X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FRO.git;a=blobdiff_plain;f=NG-RO%2Fosm_ng_ro%2Fns.py;h=a9c5ec0a894c231af940147767fde4fe265698d3;hp=eda6c487c117da0ea8cb207112d0324ec1af69fc;hb=70eeb18e4fcbb8bc3c81c88f270b59966ae4d463;hpb=14e6be87a296db26c6e6d97d0ba4d308681e9033 diff --git a/NG-RO/osm_ng_ro/ns.py b/NG-RO/osm_ng_ro/ns.py index eda6c487..a9c5ec0a 100644 --- a/NG-RO/osm_ng_ro/ns.py +++ b/NG-RO/osm_ng_ro/ns.py @@ -19,7 +19,7 @@ import logging # import yaml from traceback import format_exc as traceback_format_exc -from osm_ng_ro.ns_thread import NsWorker +from osm_ng_ro.ns_thread import NsWorker, NsWorkerException, deep_get from osm_ng_ro.validation import validate_input, deploy_schema from osm_common import dbmongo, dbmemory, fslocal, fsmongo, msglocal, msgkafka, version as common_version from osm_common.dbbase import DbException @@ -30,7 +30,7 @@ from uuid import uuid4 from threading import Lock from random import choice as random_choice from time import time -from jinja2 import Environment, Template, meta, TemplateError, TemplateNotFound, TemplateSyntaxError +from jinja2 import Environment, TemplateError, TemplateNotFound, StrictUndefined, UndefinedError from cryptography.hazmat.primitives import serialization as crypto_serialization from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.hazmat.backends import default_backend as crypto_default_backend @@ -82,10 +82,13 @@ class Ns(object): self.msg = None self.config = None # self.operations = None - self.logger = logging.getLogger("ro.ns") + self.logger = None + # ^ Getting logger inside method self.start because parent logger (ro) is not available yet. + # If done now it will not be linked to parent not getting its handler and level self.map_topic = {} self.write_lock = None self.assignment = {} + self.assignment_list = [] self.next_worker = 0 self.plugins = {} self.workers = [] @@ -102,6 +105,7 @@ class Ns(object): """ self.config = config self.config["process_id"] = get_process_id() # used for HA identity + self.logger = logging.getLogger("ro.ns") # check right version of common if versiontuple(common_version) < versiontuple(min_common_version): raise NsException("Not compatible osm/common version '{}'. Needed '{}' or higher".format( @@ -125,6 +129,8 @@ class Ns(object): elif config["storage"]["driver"] == "mongo": self.fs = fsmongo.FsMongo() self.fs.fs_connect(config["storage"]) + elif config["storage"]["driver"] is None: + pass else: raise NsException("Invalid configuration param '{}' at '[storage]':'driver'".format( config["storage"]["driver"])) @@ -159,19 +165,56 @@ class Ns(object): for worker in self.workers: worker.insert_task(("terminate",)) - def _create_worker(self, vim_account_id): - # TODO make use of the limit self.config["global"]["server.ns_threads"] + def _create_worker(self, target_id, load=True): + # Look for a thread not alive worker_id = next((i for i in range(len(self.workers)) if not self.workers[i].is_alive()), None) - if worker_id is None: - worker_id = len(self.workers) - self.workers.append(NsWorker(worker_id, self.config, self.plugins, self.db)) + if worker_id: + # re-start worker self.workers[worker_id].start() - self.workers[worker_id].insert_task(("load_vim", vim_account_id)) + else: + worker_id = len(self.workers) + if worker_id < self.config["global"]["server.ns_threads"]: + # create a new worker + self.workers.append(NsWorker(worker_id, self.config, self.plugins, self.db)) + self.workers[worker_id].start() + else: + # reached maximum number of threads, assign VIM to an existing one + worker_id = self.next_worker + self.next_worker = (self.next_worker + 1) % self.config["global"]["server.ns_threads"] + if load: + self.workers[worker_id].insert_task(("load_vim", target_id)) return worker_id - def _assign_vim(self, vim_account_id): - if vim_account_id not in self.assignment: - self.assignment[vim_account_id] = self._create_worker(vim_account_id) + def assign_vim(self, target_id): + if target_id not in self.assignment: + self.assignment[target_id] = self._create_worker(target_id) + self.assignment_list.append(target_id) + + def reload_vim(self, target_id): + # send reload_vim to the thread working with this VIM and inform all that a VIM has been changed, + # this is because database VIM information is cached for threads working with SDN + # if target_id in self.assignment: + # worker_id = self.assignment[target_id] + # self.workers[worker_id].insert_task(("reload_vim", target_id)) + for worker in self.workers: + if worker.is_alive(): + worker.insert_task(("reload_vim", target_id)) + + def unload_vim(self, target_id): + if target_id in self.assignment: + worker_id = self.assignment[target_id] + self.workers[worker_id].insert_task(("unload_vim", target_id)) + del self.assignment[target_id] + self.assignment_list.remove(target_id) + + def check_vim(self, target_id): + if target_id in self.assignment: + worker_id = self.assignment[target_id] + else: + worker_id = self._create_worker(target_id, load=False) + + worker = self.workers[worker_id] + worker.insert_task(("check_vim", target_id)) def _get_cloud_init(self, where): """ @@ -185,6 +228,9 @@ class Ns(object): if _type == "file": base_folder = vnfd["_admin"]["storage"] cloud_init_file = "{}/{}/cloud_init/{}".format(base_folder["folder"], base_folder["pkg-dir"], name) + if not self.fs: + raise NsException("Cannot read file '{}'. Filesystem not loaded, change configuration at storage.driver" + .format(cloud_init_file)) with self.fs.file_open(cloud_init_file, "r") as ci_file: cloud_init_content = ci_file.read() elif _type == "vdu": @@ -194,20 +240,16 @@ class Ns(object): return cloud_init_content def _parse_jinja2(self, cloud_init_content, params, context): + try: - env = Environment() - ast = env.parse(cloud_init_content) - mandatory_vars = meta.find_undeclared_variables(ast) - if mandatory_vars: - for var in mandatory_vars: - if not params or var not in params: - raise NsException( - "Variable '{}' defined at vnfd='{}' must be provided in the instantiation parameters" - "inside the 'additionalParamsForVnf' block".format(var, context)) - template = Template(cloud_init_content) + env = Environment(undefined=StrictUndefined) + template = env.from_string(cloud_init_content) return template.render(params or {}) - - except (TemplateError, TemplateNotFound, TemplateSyntaxError) as e: + except UndefinedError as e: + raise NsException( + "Variable '{}' defined at vnfd='{}' must be provided in the instantiation parameters" + "inside the 'additionalParamsForVnf' block".format(e, context)) + except (TemplateError, TemplateNotFound) as e: raise NsException("Error parsing Jinja2 to cloud-init content at vnfd='{}': {}".format(context, e)) def _create_db_ro_nsrs(self, nsr_id, now): @@ -226,6 +268,9 @@ class Ns(object): crypto_serialization.PublicFormat.OpenSSH ) private_key = private_key.decode('utf8') + # Change first line because Paramiko needs a explicit start with 'BEGIN RSA PRIVATE KEY' + i = private_key.find("\n") + private_key = "-----BEGIN RSA PRIVATE KEY-----" + private_key[i:] public_key = public_key.decode('utf8') except Exception as e: raise NsException("Cannot create ssh-keys: {}".format(e)) @@ -247,27 +292,24 @@ class Ns(object): return db_content def deploy(self, session, indata, version, nsr_id, *args, **kwargs): - print("ns.deploy session={} indata={} version={} nsr_id={}".format(session, indata, version, nsr_id)) + self.logger.debug("ns.deploy nsr_id={} indata={}".format(nsr_id, indata)) validate_input(indata, deploy_schema) action_id = indata.get("action_id", str(uuid4())) task_index = 0 # get current deployment - db_nsr = None - # db_nslcmop = None db_nsr_update = {} # update operation on nsrs db_vnfrs_update = {} - # db_nslcmop_update = {} # update operation on nslcmops db_vnfrs = {} # vnf's info indexed by _id - vdu2cloud_init = {} + nb_ro_tasks = 0 # for logging + vdu2cloud_init = indata.get("cloud_init_content") or {} step = '' logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id) self.logger.debug(logging_text + "Enter") try: step = "Getting ns and vnfr record from db" - # db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id}) db_nsr = self.db.get_one("nsrs", {"_id": nsr_id}) - db_ro_tasks = [] db_new_tasks = [] + tasks_by_target_record_id = {} # read from db: vnf's of this ns step = "Getting vnfrs from db" db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id}) @@ -293,12 +335,13 @@ class Ns(object): break index += 1 - def _create_task(item, action, target_record, target_record_id, extra_dict=None): + def _create_task(target_id, item, action, target_record, target_record_id, extra_dict=None): nonlocal task_index nonlocal action_id nonlocal nsr_id task = { + "target_id": target_id, # it will be removed before pushing at database "action_id": action_id, "nsr_id": nsr_id, "task_id": "{}:{}".format(action_id, task_index), @@ -313,17 +356,17 @@ class Ns(object): task_index += 1 return task - def _create_ro_task(vim_account_id, item, action, target_record, target_record_id, extra_dict=None): + def _create_ro_task(target_id, task): nonlocal action_id nonlocal task_index nonlocal now - _id = action_id + ":" + str(task_index) + _id = task["task_id"] db_ro_task = { "_id": _id, "locked_by": None, "locked_at": 0.0, - "target_id": "vim:" + vim_account_id, + "target_id": target_id, "vim_info": { "created": False, "created_items": None, @@ -336,11 +379,11 @@ class Ns(object): "modified_at": now, "created_at": now, "to_check_at": now, - "tasks": [_create_task(item, action, target_record, target_record_id, extra_dict)], + "tasks": [task], } return db_ro_task - def _process_image_params(target_image, vim_info): + def _process_image_params(target_image, vim_info, target_record_id): find_params = {} if target_image.get("image"): find_params["filter_dict"] = {"name": target_image.get("image")} @@ -350,7 +393,7 @@ class Ns(object): find_params["filter_dict"] = {"checksum": target_image.get("image_checksum")} return {"find_params": find_params} - def _process_flavor_params(target_flavor, vim_info): + def _process_flavor_params(target_flavor, vim_info, target_record_id): def _get_resource_allocation_params(quota_descriptor): """ @@ -375,9 +418,10 @@ class Ns(object): "ram": int(target_flavor["memory-mb"]), "vcpus": target_flavor["vcpu-count"], } + numa = {} + extended = {} if target_flavor.get("guest-epa"): extended = {} - numa = {} epa_vcpu_set = False if target_flavor["guest-epa"].get("numa-node-policy"): numa_node_policy = target_flavor["guest-epa"].get("numa-node-policy") @@ -438,9 +482,39 @@ class Ns(object): extra_dict["params"] = {"flavor_data": flavor_data_name} return extra_dict - def _process_net_params(target_vld, vim_info): + def _ip_profile_2_ro(ip_profile): + if not ip_profile: + return None + ro_ip_profile = { + "ip_version": "IPv4" if "v4" in ip_profile.get("ip-version", "ipv4") else "IPv6", + "subnet_address": ip_profile.get("subnet-address"), + "gateway_address": ip_profile.get("gateway-address"), + "dhcp_enabled": ip_profile["dhcp-params"].get("enabled", True), + "dhcp_start_address": ip_profile["dhcp-params"].get("start-address"), + "dhcp_count": ip_profile["dhcp-params"].get("count"), + + } + if ip_profile.get("dns-server"): + ro_ip_profile["dns_address"] = ";".join([v["address"] for v in ip_profile["dns-server"]]) + if ip_profile.get('security-group'): + ro_ip_profile["security_group"] = ip_profile['security-group'] + return ro_ip_profile + + def _process_net_params(target_vld, vim_info, target_record_id): nonlocal indata extra_dict = {} + + if vim_info.get("sdn"): + # vnf_preffix = "vnfrs:{}".format(vnfr_id) + # ns_preffix = "nsrs:{}".format(nsr_id) + vld_target_record_id, _, _ = target_record_id.rpartition(".") # remove the ending ".sdn + extra_dict["params"] = {k: vim_info[k] for k in ("sdn-ports", "target_vim", "vlds", "type") + if vim_info.get(k)} + # TODO needed to add target_id in the dependency. + if vim_info.get("target_vim"): + extra_dict["depends_on"] = [vim_info.get("target_vim") + " " + vld_target_record_id] + return extra_dict + if vim_info.get("vim_network_name"): extra_dict["find_params"] = {"filter_dict": {"name": vim_info.get("vim_network_name")}} elif vim_info.get("vim_network_id"): @@ -451,7 +525,7 @@ class Ns(object): # create extra_dict["params"] = { "net_name": "{}-{}".format(indata["name"][:16], target_vld.get("name", target_vld["id"])[:16]), - "ip_profile": vim_info.get('ip_profile'), + "ip_profile": _ip_profile_2_ro(vim_info.get('ip_profile')), "provider_network_profile": vim_info.get('provider_network'), } if not target_vld.get("underlay"): @@ -460,12 +534,13 @@ class Ns(object): extra_dict["params"]["net_type"] = "ptp" if target_vld.get("type") == "ELINE" else "data" return extra_dict - def _process_vdu_params(target_vdu, vim_info): + def _process_vdu_params(target_vdu, vim_info, target_record_id): nonlocal vnfr_id nonlocal nsr_id nonlocal indata nonlocal vnfr nonlocal vdu2cloud_init + nonlocal tasks_by_target_record_id vnf_preffix = "vnfrs:{}".format(vnfr_id) ns_preffix = "nsrs:{}".format(nsr_id) image_text = ns_preffix + ":image." + target_vdu["ns-image-id"] @@ -478,15 +553,16 @@ class Ns(object): else: net_text = vnf_preffix + ":vld." + interface["vnf-vld-id"] extra_dict["depends_on"].append(net_text) - net_item = { - "name": interface["name"], - "net_id": "TASK-" + net_text, - "vpci": interface.get("vpci"), - "type": "virtual", - # TODO mac_address: used for SR-IOV ifaces #TODO for other types - # TODO floating_ip: True/False (or it can be None) - } + net_item = {x: v for x, v in interface.items() if x in + ("name", "vpci", "port_security", "port_security_disable_strategy", "floating_ip")} + net_item["net_id"] = "TASK-" + net_text + net_item["type"] = "virtual" + # TODO mac_address: used for SR-IOV ifaces #TODO for other types + # TODO floating_ip: True/False (or it can be None) if interface.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"): + # mark the net create task as type data + if deep_get(tasks_by_target_record_id, net_text, "params", "net_type"): + tasks_by_target_record_id[net_text]["params"]["net_type"] = "data" net_item["use"] = "data" net_item["model"] = interface["type"] net_item["type"] = interface["type"] @@ -496,12 +572,15 @@ class Ns(object): else: # if interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"): net_item["use"] = "bridge" net_item["model"] = interface.get("type") + if interface.get("ip-address"): + net_item["ip_address"] = interface["ip-address"] + if interface.get("mac-address"): + net_item["mac_address"] = interface["mac-address"] net_list.append(net_item) if interface.get("mgmt-vnf"): extra_dict["mgmt_vnf_interface"] = iface_index elif interface.get("mgmt-interface"): extra_dict["mgmt_vdu_interface"] = iface_index - # cloud config cloud_config = {} if target_vdu.get("cloud-init"): @@ -536,84 +615,97 @@ class Ns(object): return extra_dict def _process_items(target_list, existing_list, db_record, db_update, db_path, item, process_params): - nonlocal db_ro_tasks nonlocal db_new_tasks + nonlocal tasks_by_target_record_id nonlocal task_index - # ensure all the target_list elements has an "id". If not assign the index + # ensure all the target_list elements has an "id". If not assign the index as id for target_index, tl in enumerate(target_list): if tl and not tl.get("id"): tl["id"] = str(target_index) - # step 1 networks to be deleted/updated - for vld_index, existing_vld in enumerate(existing_list): - target_vld = next((vld for vld in target_list if vld["id"] == existing_vld["id"]), None) - for existing_vim_index, existing_vim_info in enumerate(existing_vld.get("vim_info", ())): - if not existing_vim_info: + # step 1 items (networks,vdus,...) to be deleted/updated + for item_index, existing_item in enumerate(existing_list): + target_item = next((t for t in target_list if t["id"] == existing_item["id"]), None) + for target_vim, existing_viminfo in existing_item.get("vim_info", {}).items(): + if existing_viminfo is None: continue - if target_vld: - target_viminfo = next((target_viminfo for target_viminfo in target_vld.get("vim_info", ()) - if existing_vim_info["vim_account_id"] == target_viminfo[ - "vim_account_id"]), None) + if target_item: + target_viminfo = target_item.get("vim_info", {}).get(target_vim) else: target_viminfo = None - if not target_viminfo: + if target_viminfo is None: # must be deleted - self._assign_vim(existing_vim_info["vim_account_id"]) - db_new_tasks.append(_create_task( - item, "DELETE", - target_record="{}.{}.vim_info.{}".format(db_record, vld_index, existing_vim_index), - target_record_id="{}.{}".format(db_record, existing_vld["id"]))) + self.assign_vim(target_vim) + target_record_id = "{}.{}".format(db_record, existing_item["id"]) + item_ = item + if target_vim.startswith("sdn"): + # item must be sdn-net instead of net if target_vim is a sdn + item_ = "sdn_net" + target_record_id += ".sdn" + task = _create_task( + target_vim, item_, "DELETE", + target_record="{}.{}.vim_info.{}".format(db_record, item_index, target_vim), + target_record_id=target_record_id) + tasks_by_target_record_id[target_record_id] = task + db_new_tasks.append(task) # TODO delete # TODO check one by one the vims to be created/deleted - # step 2 networks to be created - for target_vld in target_list: - vld_index = -1 - for vld_index, existing_vld in enumerate(existing_list): - if existing_vld["id"] == target_vld["id"]: + # step 2 items (networks,vdus,...) to be created + for target_item in target_list: + item_index = -1 + for item_index, existing_item in enumerate(existing_list): + if existing_item["id"] == target_item["id"]: break else: - vld_index += 1 - db_update[db_path + ".{}".format(vld_index)] = target_vld - existing_list.append(target_vld) - existing_vld = None + item_index += 1 + db_update[db_path + ".{}".format(item_index)] = target_item + existing_list.append(target_item) + existing_item = None - for vim_index, vim_info in enumerate(target_vld["vim_info"]): + for target_vim, target_viminfo in target_item.get("vim_info", {}).items(): existing_viminfo = None - if existing_vld: - existing_viminfo = next( - (existing_viminfo for existing_viminfo in existing_vld.get("vim_info", ()) - if vim_info["vim_account_id"] == existing_viminfo["vim_account_id"]), None) + if existing_item: + existing_viminfo = existing_item.get("vim_info", {}).get(target_vim) # TODO check if different. Delete and create??? # TODO delete if not exist - if existing_viminfo: + if existing_viminfo is not None: continue - extra_dict = process_params(target_vld, vim_info) - - self._assign_vim(vim_info["vim_account_id"]) - db_ro_tasks.append(_create_ro_task( - vim_info["vim_account_id"], item, "CREATE", - target_record="{}.{}.vim_info.{}".format(db_record, vld_index, vim_index), - target_record_id="{}.{}".format(db_record, target_vld["id"]), - extra_dict=extra_dict)) - - db_update[db_path + ".{}".format(vld_index)] = target_vld + target_record_id = "{}.{}".format(db_record, target_item["id"]) + item_ = item + if target_vim.startswith("sdn"): + # item must be sdn-net instead of net if target_vim is a sdn + item_ = "sdn_net" + target_record_id += ".sdn" + extra_dict = process_params(target_item, target_viminfo, target_record_id) + + self.assign_vim(target_vim) + task = _create_task( + target_vim, item_, "CREATE", + target_record="{}.{}.vim_info.{}".format(db_record, item_index, target_vim), + target_record_id=target_record_id, + extra_dict=extra_dict) + tasks_by_target_record_id[target_record_id] = task + db_new_tasks.append(task) + if target_item.get("common_id"): + task["common_id"] = target_item["common_id"] + + db_update[db_path + ".{}".format(item_index)] = target_item def _process_action(indata): - nonlocal db_ro_tasks nonlocal db_new_tasks nonlocal task_index nonlocal db_vnfrs nonlocal db_ro_nsr - if indata["action"] == "inject_ssh_key": - key = indata.get("key") - user = indata.get("user") - password = indata.get("password") + if indata["action"]["action"] == "inject_ssh_key": + key = indata["action"].get("key") + user = indata["action"].get("user") + password = indata["action"].get("password") for vnf in indata.get("vnf", ()): - if vnf.get("_id") not in db_vnfrs: + if vnf["_id"] not in db_vnfrs: raise NsException("Invalid vnf={}".format(vnf["_id"])) db_vnfr = db_vnfrs[vnf["_id"]] for target_vdu in vnf.get("vdur", ()): @@ -621,13 +713,13 @@ class Ns(object): i_v[1]["id"] == target_vdu["id"]), (None, None)) if not vdur: raise NsException("Invalid vdu vnf={}.{}".format(vnf["_id"], target_vdu["id"])) - vim_info = vdur["vim_info"][0] - self._assign_vim(vim_info["vim_account_id"]) + target_vim, vim_info = next(k_v for k_v in vdur["vim_info"].items()) + self.assign_vim(target_vim) target_record = "vnfrs:{}:vdur.{}.ssh_keys".format(vnf["_id"], vdu_index) extra_dict = { "depends_on": ["vnfrs:{}:vdur.{}".format(vnf["_id"], vdur["id"])], "params": { - "ip_address": vdur.gt("ip_address"), + "ip_address": vdur.get("ip-address"), "user": user, "key": key, "password": password, @@ -636,10 +728,11 @@ class Ns(object): "schema_version": db_ro_nsr["_admin"]["schema_version"] } } - db_ro_tasks.append(_create_ro_task(vim_info["vim_account_id"], "vdu", "EXEC", - target_record=target_record, - target_record_id=None, - extra_dict=extra_dict)) + task = _create_task(target_vim, "vdu", "EXEC", + target_record=target_record, + target_record_id=None, + extra_dict=extra_dict) + db_new_tasks.append(task) with self.write_lock: if indata.get("action"): @@ -653,13 +746,13 @@ class Ns(object): db_path="vld", item="net", process_params=_process_net_params) step = "process NS images" - _process_items(target_list=indata["image"] or [], existing_list=db_nsr.get("image") or [], + _process_items(target_list=indata.get("image") or [], existing_list=db_nsr.get("image") or [], db_record="nsrs:{}:image".format(nsr_id), db_update=db_nsr_update, db_path="image", item="image", process_params=_process_image_params) step = "process NS flavors" - _process_items(target_list=indata["flavor"] or [], existing_list=db_nsr.get("flavor") or [], + _process_items(target_list=indata.get("flavor") or [], existing_list=db_nsr.get("flavor") or [], db_record="nsrs:{}:flavor".format(nsr_id), db_update=db_nsr_update, db_path="flavor", item="flavor", process_params=_process_flavor_params) @@ -681,17 +774,27 @@ class Ns(object): db_update=db_vnfrs_update[vnfr["_id"]], db_path="vdur", item="vdu", process_params=_process_vdu_params) - step = "Updating database, Creating ro_tasks" - if db_ro_tasks: - self.db.create_list("ro_tasks", db_ro_tasks) - step = "Updating database, Appending tasks to ro_tasks" - for task in db_new_tasks: - if not self.db.set_one("ro_tasks", q_filter={"tasks.target_record": task["target_record"]}, + for db_task in db_new_tasks: + step = "Updating database, Appending tasks to ro_tasks" + target_id = db_task.pop("target_id") + common_id = db_task.get("common_id") + if common_id: + if self.db.set_one("ro_tasks", + q_filter={"target_id": target_id, + "tasks.common_id": common_id}, update_dict={"to_check_at": now, "modified_at": now}, - push={"tasks": task}, fail_on_empty=False): - self.logger.error(logging_text + "Cannot find task for target_record={}". - format(task["target_record"])) - # TODO something else appart from logging? + push={"tasks": db_task}, fail_on_empty=False): + continue + if not self.db.set_one("ro_tasks", + q_filter={"target_id": target_id, + "tasks.target_record": db_task["target_record"]}, + update_dict={"to_check_at": now, "modified_at": now}, + push={"tasks": db_task}, fail_on_empty=False): + # Create a ro_task + step = "Updating database, Creating ro_tasks" + db_ro_task = _create_ro_task(target_id, db_task) + nb_ro_tasks += 1 + self.db.create("ro_tasks", db_ro_task) step = "Updating database, nsrs" if db_nsr_update: self.db.set_one("nsrs", {"_id": nsr_id}, db_nsr_update) @@ -700,7 +803,8 @@ class Ns(object): step = "Updating database, vnfrs={}".format(vnfr_id) self.db.set_one("vnfrs", {"_id": vnfr_id}, db_vnfr_update) - self.logger.debug(logging_text + "Exit") + self.logger.debug(logging_text + "Exit. Created {} ro_tasks; {} tasks".format(nb_ro_tasks, + len(db_new_tasks))) return {"status": "ok", "nsr_id": nsr_id, "action_id": action_id}, action_id, True except Exception as e: @@ -712,50 +816,18 @@ class Ns(object): raise NsException(e) def delete(self, session, indata, version, nsr_id, *args, **kwargs): - print("ns.delete session={} indata={} version={} nsr_id={}".format(session, indata, version, nsr_id)) - # TODO del when ALL "tasks.nsr_id" are None of nsr_id + self.logger.debug("ns.delete version={} nsr_id={}".format(version, nsr_id)) # self.db.del_list({"_id": ro_task["_id"], "tasks.nsr_id.ne": nsr_id}) - retries = 5 - for retry in range(retries): - with self.write_lock: - ro_tasks = self.db.get_list("ro_tasks", {"tasks.nsr_id": nsr_id}) - if not ro_tasks: - break - now = time() - conflict = False - for ro_task in ro_tasks: - db_update = {} - to_delete = True - for index, task in enumerate(ro_task["tasks"]): - if not task: - pass - elif task["nsr_id"] == nsr_id: - db_update["tasks.{}".format(index)] = None - else: - to_delete = False # used by other nsr, cannot be deleted - # delete or update if nobody has changed ro_task meanwhile. Used modified_at for known if changed - if to_delete: - if not self.db.del_one("ro_tasks", - q_filter={"_id": ro_task["_id"], "modified_at": ro_task["modified_at"]}, - fail_on_empty=False): - conflict = True - elif db_update: - db_update["modified_at"] = now - if not self.db.set_one("ro_tasks", - q_filter={"_id": ro_task["_id"], "modified_at": ro_task["modified_at"]}, - update_dict=db_update, - fail_on_empty=False): - conflict = True - if not conflict: - break - else: - raise NsException("Exceeded {} retries".format(retries)) - + with self.write_lock: + try: + NsWorker.delete_db_tasks(self.db, nsr_id, None) + except NsWorkerException as e: + raise NsException(e) return None, None, True def status(self, session, indata, version, nsr_id, action_id, *args, **kwargs): - print("ns.status session={} indata={} version={} nsr_id={}, action_id={}".format(session, indata, version, - nsr_id, action_id)) + # self.logger.debug("ns.status version={} nsr_id={}, action_id={} indata={}" + # .format(version, nsr_id, action_id, indata)) task_list = [] done = 0 total = 0 @@ -764,12 +836,14 @@ class Ns(object): details = [] for ro_task in ro_tasks: for task in ro_task["tasks"]: - if task["action_id"] == action_id: + if task and task["action_id"] == action_id: task_list.append(task) total += 1 if task["status"] == "FAILED": global_status = "FAILED" - details.append(ro_task.get("vim_details", '')) + error_text = "Error at {} {}: {}".format(task["action"].lower(), task["item"], + ro_task["vim_info"].get("vim_details") or "unknown") + details.append(error_text) elif task["status"] in ("SCHEDULED", "BUILD"): if global_status != "FAILED": global_status = "BUILD"