#
##
-""""
+"""
This is thread that interacts with a VIM. It processes TASKs sequentially against a single VIM.
The tasks are stored at database in table ro_tasks
A single ro_task refers to a VIM element (flavor, image, network, ...).
:param target_dict: dictionary to be read
:param args: list of keys to read from target_dict
:param kwargs: only can contain default=value to return if key is not present in the nested dictionary
- :return: The wanted value if exist, None or default otherwise
+ :return: The wanted value if exists, None or default otherwise
"""
for key in args:
if not isinstance(target_dict, dict) or key not in target_dict:
ro_task["target_id"],
vim_id,
ro_vim_item_update.get("vim_status"),
- ro_vim_item_update.get("vim_message")
- if ro_vim_item_update.get("vim_status") != "ACTIVE"
- else "",
+ (
+ ro_vim_item_update.get("vim_message")
+ if ro_vim_item_update.get("vim_status") != "ACTIVE"
+ else ""
+ ),
)
)
return "DONE", ro_vim_item_update_ok
+class VimInteractionClassification(VimInteractionBase):
+ def new(self, ro_task, task_index, task_depends):
+ task = ro_task["tasks"][task_index]
+ task_id = task["task_id"]
+ created = False
+ target_vim = self.my_vims[ro_task["target_id"]]
+
+ try:
+ created = True
+ params = task["params"]
+ params_copy = deepcopy(params)
+
+ name = params_copy.pop("name")
+ logical_source_port_index = int(
+ params_copy.pop("logical_source_port_index")
+ )
+ logical_source_port = params_copy["logical_source_port"]
+
+ if logical_source_port.startswith("TASK-"):
+ vm_id = task_depends[logical_source_port]
+ params_copy["logical_source_port"] = target_vim.refresh_vms_status(
+ [vm_id]
+ )[vm_id]["interfaces"][logical_source_port_index]["vim_interface_id"]
+
+ vim_classification_id = target_vim.new_classification(
+ name, "legacy_flow_classifier", params_copy
+ )
+
+ ro_vim_item_update = {
+ "vim_id": vim_classification_id,
+ "vim_status": "DONE",
+ "created": created,
+ "vim_details": None,
+ "vim_message": None,
+ }
+ self.logger.debug(
+ "task={} {} created={}".format(task_id, ro_task["target_id"], created)
+ )
+
+ return "DONE", ro_vim_item_update
+ except (vimconn.VimConnException, NsWorkerException) as e:
+ self.logger.debug(traceback.format_exc())
+ self.logger.error(
+ "task={} {} new-vm: {}".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
+
+ def delete(self, ro_task, task_index):
+ task = ro_task["tasks"][task_index]
+ task_id = task["task_id"]
+ classification_vim_id = ro_task["vim_info"]["vim_id"]
+ ro_vim_item_update_ok = {
+ "vim_status": "DELETED",
+ "created": False,
+ "vim_message": "DELETED",
+ "vim_id": None,
+ }
+
+ try:
+ if classification_vim_id:
+ target_vim = self.my_vims[ro_task["target_id"]]
+ target_vim.delete_classification(classification_vim_id)
+ except vimconn.VimConnNotFoundException:
+ ro_vim_item_update_ok["vim_message"] = "already deleted"
+ except vimconn.VimConnException as e:
+ self.logger.error(
+ "ro_task={} vim={} del-classification={}: {}".format(
+ ro_task["_id"], ro_task["target_id"], classification_vim_id, e
+ )
+ )
+ ro_vim_item_update = {
+ "vim_status": "VIM_ERROR",
+ "vim_message": "Error while deleting: {}".format(e),
+ }
+
+ return "FAILED", ro_vim_item_update
+
+ self.logger.debug(
+ "task={} {} del-classification={} {}".format(
+ task_id,
+ ro_task["target_id"],
+ classification_vim_id,
+ ro_vim_item_update_ok.get("vim_message", ""),
+ )
+ )
+
+ return "DONE", ro_vim_item_update_ok
+
+
+class VimInteractionSfi(VimInteractionBase):
+ def new(self, ro_task, task_index, task_depends):
+ task = ro_task["tasks"][task_index]
+ task_id = task["task_id"]
+ created = False
+ target_vim = self.my_vims[ro_task["target_id"]]
+
+ try:
+ created = True
+ params = task["params"]
+ params_copy = deepcopy(params)
+ name = params_copy["name"]
+ ingress_port = params_copy["ingress_port"]
+ egress_port = params_copy["egress_port"]
+ ingress_port_index = params_copy["ingress_port_index"]
+ egress_port_index = params_copy["egress_port_index"]
+
+ ingress_port_id = ingress_port
+ egress_port_id = egress_port
+
+ vm_id = task_depends[ingress_port]
+
+ if ingress_port.startswith("TASK-"):
+ ingress_port_id = target_vim.refresh_vms_status([vm_id])[vm_id][
+ "interfaces"
+ ][ingress_port_index]["vim_interface_id"]
+
+ if ingress_port == egress_port:
+ egress_port_id = ingress_port_id
+ else:
+ if egress_port.startswith("TASK-"):
+ egress_port_id = target_vim.refresh_vms_status([vm_id])[vm_id][
+ "interfaces"
+ ][egress_port_index]["vim_interface_id"]
+
+ ingress_port_id_list = [ingress_port_id]
+ egress_port_id_list = [egress_port_id]
+
+ vim_sfi_id = target_vim.new_sfi(
+ name, ingress_port_id_list, egress_port_id_list, sfc_encap=False
+ )
+
+ ro_vim_item_update = {
+ "vim_id": vim_sfi_id,
+ "vim_status": "DONE",
+ "created": created,
+ "vim_details": None,
+ "vim_message": None,
+ }
+ self.logger.debug(
+ "task={} {} created={}".format(task_id, ro_task["target_id"], created)
+ )
+
+ return "DONE", ro_vim_item_update
+ except (vimconn.VimConnException, NsWorkerException) as e:
+ self.logger.debug(traceback.format_exc())
+ self.logger.error(
+ "task={} {} new-vm: {}".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
+
+ def delete(self, ro_task, task_index):
+ task = ro_task["tasks"][task_index]
+ task_id = task["task_id"]
+ sfi_vim_id = ro_task["vim_info"]["vim_id"]
+ ro_vim_item_update_ok = {
+ "vim_status": "DELETED",
+ "created": False,
+ "vim_message": "DELETED",
+ "vim_id": None,
+ }
+
+ try:
+ if sfi_vim_id:
+ target_vim = self.my_vims[ro_task["target_id"]]
+ target_vim.delete_sfi(sfi_vim_id)
+ except vimconn.VimConnNotFoundException:
+ ro_vim_item_update_ok["vim_message"] = "already deleted"
+ except vimconn.VimConnException as e:
+ self.logger.error(
+ "ro_task={} vim={} del-sfi={}: {}".format(
+ ro_task["_id"], ro_task["target_id"], sfi_vim_id, e
+ )
+ )
+ ro_vim_item_update = {
+ "vim_status": "VIM_ERROR",
+ "vim_message": "Error while deleting: {}".format(e),
+ }
+
+ return "FAILED", ro_vim_item_update
+
+ self.logger.debug(
+ "task={} {} del-sfi={} {}".format(
+ task_id,
+ ro_task["target_id"],
+ sfi_vim_id,
+ ro_vim_item_update_ok.get("vim_message", ""),
+ )
+ )
+
+ return "DONE", ro_vim_item_update_ok
+
+
+class VimInteractionSf(VimInteractionBase):
+ def new(self, ro_task, task_index, task_depends):
+ task = ro_task["tasks"][task_index]
+ task_id = task["task_id"]
+ created = False
+ target_vim = self.my_vims[ro_task["target_id"]]
+
+ try:
+ created = True
+ params = task["params"]
+ params_copy = deepcopy(params)
+ name = params_copy["name"]
+ sfi_list = params_copy["sfis"]
+ sfi_id_list = []
+
+ for sfi in sfi_list:
+ sfi_id = task_depends[sfi] if sfi.startswith("TASK-") else sfi
+ sfi_id_list.append(sfi_id)
+
+ vim_sf_id = target_vim.new_sf(name, sfi_id_list, sfc_encap=False)
+
+ ro_vim_item_update = {
+ "vim_id": vim_sf_id,
+ "vim_status": "DONE",
+ "created": created,
+ "vim_details": None,
+ "vim_message": None,
+ }
+ self.logger.debug(
+ "task={} {} created={}".format(task_id, ro_task["target_id"], created)
+ )
+
+ return "DONE", ro_vim_item_update
+ except (vimconn.VimConnException, NsWorkerException) as e:
+ self.logger.debug(traceback.format_exc())
+ self.logger.error(
+ "task={} {} new-vm: {}".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
+
+ def delete(self, ro_task, task_index):
+ task = ro_task["tasks"][task_index]
+ task_id = task["task_id"]
+ sf_vim_id = ro_task["vim_info"]["vim_id"]
+ ro_vim_item_update_ok = {
+ "vim_status": "DELETED",
+ "created": False,
+ "vim_message": "DELETED",
+ "vim_id": None,
+ }
+
+ try:
+ if sf_vim_id:
+ target_vim = self.my_vims[ro_task["target_id"]]
+ target_vim.delete_sf(sf_vim_id)
+ except vimconn.VimConnNotFoundException:
+ ro_vim_item_update_ok["vim_message"] = "already deleted"
+ except vimconn.VimConnException as e:
+ self.logger.error(
+ "ro_task={} vim={} del-sf={}: {}".format(
+ ro_task["_id"], ro_task["target_id"], sf_vim_id, e
+ )
+ )
+ ro_vim_item_update = {
+ "vim_status": "VIM_ERROR",
+ "vim_message": "Error while deleting: {}".format(e),
+ }
+
+ return "FAILED", ro_vim_item_update
+
+ self.logger.debug(
+ "task={} {} del-sf={} {}".format(
+ task_id,
+ ro_task["target_id"],
+ sf_vim_id,
+ ro_vim_item_update_ok.get("vim_message", ""),
+ )
+ )
+
+ return "DONE", ro_vim_item_update_ok
+
+
+class VimInteractionSfp(VimInteractionBase):
+ def new(self, ro_task, task_index, task_depends):
+ task = ro_task["tasks"][task_index]
+ task_id = task["task_id"]
+ created = False
+ target_vim = self.my_vims[ro_task["target_id"]]
+
+ try:
+ created = True
+ params = task["params"]
+ params_copy = deepcopy(params)
+ name = params_copy["name"]
+ sf_list = params_copy["sfs"]
+ classification_list = params_copy["classifications"]
+
+ classification_id_list = []
+ sf_id_list = []
+
+ for classification in classification_list:
+ classi_id = (
+ task_depends[classification]
+ if classification.startswith("TASK-")
+ else classification
+ )
+ classification_id_list.append(classi_id)
+
+ for sf in sf_list:
+ sf_id = task_depends[sf] if sf.startswith("TASK-") else sf
+ sf_id_list.append(sf_id)
+
+ vim_sfp_id = target_vim.new_sfp(
+ name, classification_id_list, sf_id_list, sfc_encap=False
+ )
+
+ ro_vim_item_update = {
+ "vim_id": vim_sfp_id,
+ "vim_status": "DONE",
+ "created": created,
+ "vim_details": None,
+ "vim_message": None,
+ }
+ self.logger.debug(
+ "task={} {} created={}".format(task_id, ro_task["target_id"], created)
+ )
+
+ return "DONE", ro_vim_item_update
+ except (vimconn.VimConnException, NsWorkerException) as e:
+ self.logger.debug(traceback.format_exc())
+ self.logger.error(
+ "task={} {} new-vm: {}".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
+
+ def delete(self, ro_task, task_index):
+ task = ro_task["tasks"][task_index]
+ task_id = task["task_id"]
+ sfp_vim_id = ro_task["vim_info"]["vim_id"]
+ ro_vim_item_update_ok = {
+ "vim_status": "DELETED",
+ "created": False,
+ "vim_message": "DELETED",
+ "vim_id": None,
+ }
+
+ try:
+ if sfp_vim_id:
+ target_vim = self.my_vims[ro_task["target_id"]]
+ target_vim.delete_sfp(sfp_vim_id)
+ except vimconn.VimConnNotFoundException:
+ ro_vim_item_update_ok["vim_message"] = "already deleted"
+ except vimconn.VimConnException as e:
+ self.logger.error(
+ "ro_task={} vim={} del-sfp={}: {}".format(
+ ro_task["_id"], ro_task["target_id"], sfp_vim_id, e
+ )
+ )
+ ro_vim_item_update = {
+ "vim_status": "VIM_ERROR",
+ "vim_message": "Error while deleting: {}".format(e),
+ }
+
+ return "FAILED", ro_vim_item_update
+
+ self.logger.debug(
+ "task={} {} del-sfp={} {}".format(
+ task_id,
+ ro_task["target_id"],
+ sfp_vim_id,
+ ro_vim_item_update_ok.get("vim_message", ""),
+ )
+ )
+
+ return "DONE", ro_vim_item_update_ok
+
+
class VimInteractionVdu(VimInteractionBase):
max_retries_inject_ssh_key = 20 # 20 times
time_retries_inject_ssh_key = 30 # wevery 30 seconds
task = ro_task["tasks"][task_index]
task_id = task["task_id"]
created = False
- created_items = {}
target_vim = self.my_vims[ro_task["target_id"]]
-
try:
created = True
params = task["params"]
)
affinity_group["affinity_group_id"] = affinity_group_id
-
vim_vm_id, created_items = target_vim.new_vminstance(**params_copy)
interfaces = [iface["vim_id"] for iface in params_copy["net_list"]]
ro_task["target_id"],
vim_id,
ro_vim_item_update.get("vim_status"),
- ro_vim_item_update.get("vim_message")
- if ro_vim_item_update.get("vim_status") != "ACTIVE"
- else "",
+ (
+ ro_vim_item_update.get("vim_message")
+ if ro_vim_item_update.get("vim_status") != "ACTIVE"
+ else ""
+ ),
)
)
try:
# FIND
+ vim_image_id = ""
if task.get("find_params"):
- vim_images = target_vim.get_image_list(**task["find_params"])
+ vim_images = target_vim.get_image_list(
+ task["find_params"].get("filter_dict", {})
+ )
if not vim_images:
raise NsWorkerExceptionNotFound(
ro_vim_item_update = {
"vim_id": vim_image_id,
- "vim_status": "DONE",
+ "vim_status": "ACTIVE",
"created": created,
"created_items": created_items,
"vim_details": None,
return "FAILED", ro_vim_item_update
+class VimInteractionSharedVolume(VimInteractionBase):
+ def delete(self, ro_task, task_index):
+ task = ro_task["tasks"][task_index]
+ task_id = task["task_id"]
+ shared_volume_vim_id = ro_task["vim_info"]["vim_id"]
+ created_items = ro_task["vim_info"]["created_items"]
+ ro_vim_item_update_ok = {
+ "vim_status": "DELETED",
+ "created": False,
+ "vim_message": "DELETED",
+ "vim_id": None,
+ }
+ if created_items and created_items.get(shared_volume_vim_id).get("keep"):
+ ro_vim_item_update_ok = {
+ "vim_status": "ACTIVE",
+ "created": False,
+ "vim_message": None,
+ }
+ return "DONE", ro_vim_item_update_ok
+ try:
+ if shared_volume_vim_id:
+ target_vim = self.my_vims[ro_task["target_id"]]
+ target_vim.delete_shared_volumes(shared_volume_vim_id)
+ except vimconn.VimConnNotFoundException:
+ ro_vim_item_update_ok["vim_message"] = "already deleted"
+ except vimconn.VimConnException as e:
+ self.logger.error(
+ "ro_task={} vim={} del-shared-volume={}: {}".format(
+ ro_task["_id"], ro_task["target_id"], shared_volume_vim_id, e
+ )
+ )
+ ro_vim_item_update = {
+ "vim_status": "VIM_ERROR",
+ "vim_message": "Error while deleting: {}".format(e),
+ }
+
+ return "FAILED", ro_vim_item_update
+
+ self.logger.debug(
+ "task={} {} del-shared-volume={} {}".format(
+ task_id,
+ ro_task["target_id"],
+ shared_volume_vim_id,
+ ro_vim_item_update_ok.get("vim_message", ""),
+ )
+ )
+
+ return "DONE", ro_vim_item_update_ok
+
+ def new(self, ro_task, task_index, task_depends):
+ task = ro_task["tasks"][task_index]
+ task_id = task["task_id"]
+ created = False
+ created_items = {}
+ target_vim = self.my_vims[ro_task["target_id"]]
+
+ try:
+ shared_volume_vim_id = None
+ shared_volume_data = None
+
+ if task.get("params"):
+ shared_volume_data = task["params"]
+
+ if shared_volume_data:
+ self.logger.info(
+ f"Creating the new shared_volume for {shared_volume_data}\n"
+ )
+ (
+ shared_volume_name,
+ shared_volume_vim_id,
+ ) = target_vim.new_shared_volumes(shared_volume_data)
+ created = True
+ created_items[shared_volume_vim_id] = {
+ "name": shared_volume_name,
+ "keep": shared_volume_data.get("keep"),
+ }
+
+ ro_vim_item_update = {
+ "vim_id": shared_volume_vim_id,
+ "vim_status": "ACTIVE",
+ "created": created,
+ "created_items": created_items,
+ "vim_details": None,
+ "vim_message": None,
+ }
+ self.logger.debug(
+ "task={} {} new-shared-volume={} created={}".format(
+ task_id, ro_task["target_id"], shared_volume_vim_id, created
+ )
+ )
+
+ return "DONE", ro_vim_item_update
+ except (vimconn.VimConnException, NsWorkerException) as e:
+ self.logger.error(
+ "task={} vim={} new-shared-volume:"
+ " {}".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
+
+
class VimInteractionFlavor(VimInteractionBase):
def delete(self, ro_task, task_index):
task = ro_task["tasks"][task_index]
created = False
created_items = {}
target_vim = self.my_vims[ro_task["target_id"]]
-
try:
# FIND
vim_flavor_id = None
- if task.get("find_params"):
+ if task.get("find_params", {}).get("vim_flavor_id"):
+ vim_flavor_id = task["find_params"]["vim_flavor_id"]
+ elif task.get("find_params", {}).get("flavor_data"):
try:
flavor_data = task["find_params"]["flavor_data"]
vim_flavor_id = target_vim.get_flavor_id_from_data(flavor_data)
ro_vim_item_update = {
"vim_id": vim_flavor_id,
- "vim_status": "DONE",
+ "vim_status": "ACTIVE",
"created": created,
"created_items": created_items,
"vim_details": None,
try:
affinity_group_vim_id = None
affinity_group_data = None
+ param_affinity_group_id = ""
if task.get("params"):
affinity_group_data = task["params"].get("affinity_group_data")
ro_vim_item_update = {
"vim_id": affinity_group_vim_id,
- "vim_status": "DONE",
+ "vim_status": "ACTIVE",
"created": created,
"created_items": created_items,
"vim_details": None,
target_vim = self.my_vims[ro_task["target_id"]]
try:
+ vim_vm_id = ""
if task.get("params"):
vim_vm_id = task["params"].get("vim_vm_id")
action = task["params"].get("action")
# created = True
ro_vim_item_update = {
"vim_id": vim_vm_id,
- "vim_status": "DONE",
+ "vim_status": "ACTIVE",
"created": created,
"created_items": created_items,
"vim_details": None,
@staticmethod
def _match_pci(port_pci, mapping):
"""
- Check if port_pci matches with mapping
- mapping can have brackets to indicate that several chars are accepted. e.g
+ Check if port_pci matches with mapping.
+ The mapping can have brackets to indicate that several chars are accepted. e.g
pci '0000:af:10.1' matches with '0000:af:1[01].[1357]'
:param port_pci: text
:param mapping: text, can contain brackets to indicate several chars are available
try:
# CREATE
+ db_vim = {}
params = task["params"]
vlds_to_connect = params.get("vlds", [])
associated_vim = params.get("target_vim")
new_port = {
"service_endpoint_id": pmap.get("service_endpoint_id")
or service_endpoint_id,
- "service_endpoint_encapsulation_type": "dot1q"
- if port["type"] == "SR-IOV"
- else None,
+ "service_endpoint_encapsulation_type": (
+ "dot1q" if port["type"] == "SR-IOV" else None
+ ),
"service_endpoint_encapsulation_info": {
"vlan": port.get("vlan"),
"mac": port.get("mac-address"),
refreshed_vim_info = {}
try:
+ vim_vm_id = ""
if task.get("params"):
vim_vm_id = task["params"].get("vim_vm_id")
migrate_host = task["params"].get("migrate_host")
target_vim = self.my_vims[ro_task["target_id"]]
try:
+ params = task["params"]
+ params_copy = deepcopy(params)
+ target_flavor_uuid = task_depends[params_copy["flavor_id"]]
+ vim_vm_id = ""
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))
+ self.logger.info("vim_vm_id %s", vim_vm_id)
if target_flavor_uuid is not None:
resized_status = target_vim.resize_instance(
ro_vim_item_update = {
"vim_id": vim_vm_id,
- "vim_status": "DONE",
+ "vim_status": "ACTIVE",
"created": created,
"created_items": created_items,
"vim_details": None,
self.db = db
self.item2class = {
"net": VimInteractionNet(self.db, self.my_vims, self.db_vims, self.logger),
+ "shared-volumes": VimInteractionSharedVolume(
+ self.db, self.my_vims, self.db_vims, self.logger
+ ),
+ "classification": VimInteractionClassification(
+ self.db, self.my_vims, self.db_vims, self.logger
+ ),
+ "sfi": VimInteractionSfi(self.db, self.my_vims, self.db_vims, self.logger),
+ "sf": VimInteractionSf(self.db, self.my_vims, self.db_vims, self.logger),
+ "sfp": VimInteractionSfp(self.db, self.my_vims, self.db_vims, self.logger),
"vdu": VimInteractionVdu(self.db, self.my_vims, self.db_vims, self.logger),
"image": VimInteractionImage(
self.db, self.my_vims, self.db_vims, self.logger
target_database = (
"vim_accounts"
if target == "vim"
- else "wim_accounts"
- if target == "wim"
- else "sdns"
+ else "wim_accounts" if target == "wim" else "sdns"
)
+ error_text = ""
try:
step = "Getting {} from db".format(target_id)
"""
Load or reload a vim_account, sdn_controller or wim_account.
Read content from database, load the plugin if not loaded.
- In case of error loading the plugin, it load a failing VIM_connector
+ In case of error loading the plugin, it loads a failing VIM_connector
It fills self db_vims dictionary, my_vims dictionary and vim_targets list
:param target_id: Contains type:_id; where type can be 'vim', ...
:return: None if ok, descriptive text if error
target_database = (
"vim_accounts"
if target == "vim"
- else "wim_accounts"
- if target == "wim"
- else "sdns"
+ else "wim_accounts" if target == "wim" else "sdns"
)
plugin_name = ""
vim = None
and task["action"] == "CREATE"
):
# set to finished
- db_update["tasks.{}.status".format(index)] = task[
- "status"
- ] = "FINISHED"
+ db_update["tasks.{}.status".format(index)] = task["status"] = (
+ "FINISHED"
+ )
elif task["action"] == "CREATE" and task["status"] not in (
"FINISHED",
"SUPERSEDED",
"""
my_task = ro_task["tasks"][task_index]
task_id = my_task["task_id"]
- task_status = None
if my_task["status"] == "FAILED":
return None, None # TODO need to be retry??
task_path = "tasks.{}.status".format(task_index)
try:
db_vim_info_update = None
+ dependency_ro_task = {}
if task["status"] == "SCHEDULED":
# check if tasks that this depends on have been completed
task_depends[dependency_task_id] = dependency_ro_task[
"vim_info"
]["vim_id"]
- task_depends[
- "TASK-{}".format(dependency_task_id)
- ] = dependency_ro_task["vim_info"]["vim_id"]
+ task_depends["TASK-{}".format(dependency_task_id)] = (
+ dependency_ro_task["vim_info"]["vim_id"]
+ )
if dependency_not_completed:
self.logger.warning(
lock_object = LockRenew.add_lock_object(
"ro_tasks", ro_task, self
)
-
if task["action"] == "DELETE":
(
new_status,
lock_object["locked_at"],
lock_object["locked_at"] + self.task_locked_time,
]
- # locked_at contains two times to avoid race condition. In case the lock has been renew, it will
+ # locked_at contains two times to avoid race condition. In case the lock has been renewed, it will
# contain exactly locked_at + self.task_locked_time
LockRenew.remove_lock_object(lock_object)
# modify own task. Try filtering by to_next_check. For race condition if to_check_at has been modified,
# outside this task (by ro_nbi) do not update it
db_ro_task_update["locked_by"] = None
- # locked_at converted to int only for debugging. When has not decimals it means it has been unlocked
+ # locked_at converted to int only for debugging. When it is not decimals it means it has been unlocked
db_ro_task_update["locked_at"] = int(now - self.task_locked_time)
db_ro_task_update["modified_at"] = now
db_ro_task_update["to_check_at"] = next_check_at
projects / osm / RO.git / blobdiff