+ self.refresh_tasks = []
+ """Contains time ordered task list for refreshing the status of VIM VMs and nets"""
+
+ self.pending_tasks = []
+ """Contains time ordered task list for creation, deletion of VIM VMs and nets"""
+
+ self.grouped_tasks = {}
+ """ It contains all the creation/deletion pending tasks grouped by its concrete vm, net, etc
+ <item><item_id>:
+ - <task1> # e.g. CREATE task
+ <task2> # e.g. DELETE task
+ """
+
+ def _reload_vim_actions(self):
+ """
+ Read actions from database and reload them at memory. Fill self.refresh_list, pending_list, vim_actions
+ :return: None
+ """
+ try:
+ action_completed = False
+ task_list = []
+ old_action_key = None
+
+ old_item_id = ""
+ old_item = ""
+ old_created_at = 0.0
+ database_limit = 200
+ while True:
+ # get 200 (database_limit) entries each time
+ with self.db_lock:
+ vim_actions = self.db.get_rows(FROM="vim_actions",
+ WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
+ "item_id>=": old_item_id},
+ ORDER_BY=("item_id", "item", "created_at",),
+ LIMIT=database_limit)
+ for task in vim_actions:
+ item = task["item"]
+ item_id = task["item_id"]
+
+ # skip the first entries that are already processed in the previous pool of 200
+ if old_item_id:
+ if item_id == old_item_id and item == old_item and task["created_at"] == old_created_at:
+ old_item_id = False # next one will be a new un-processed task
+ continue
+
+ action_key = item + item_id
+ if old_action_key != action_key:
+ if not action_completed and task_list:
+ # This will fill needed task parameters into memory, and insert the task if needed in
+ # self.pending_tasks or self.refresh_tasks
+ try:
+ self._insert_pending_tasks(task_list)
+ except Exception as e:
+ self.logger.critical(
+ "Unexpected exception at _reload_vim_actions:_insert_pending_tasks: " + str(e),
+ exc_info=True)
+ task_list = []
+ old_action_key = action_key
+ action_completed = False
+ elif action_completed:
+ continue
+
+ if task["status"] == "SCHEDULED" or task["action"] == "CREATE" or task["action"] == "FIND":
+ task_list.append(task)
+ elif task["action"] == "DELETE":
+ # action completed because deleted and status is not SCHEDULED. Not needed anything
+ action_completed = True
+ if len(vim_actions) == database_limit:
+ # update variables for get the next database iteration
+ old_item_id = item_id
+ old_item = item
+ old_created_at = task["created_at"]
+ else:
+ break
+ # Last actions group need to be inserted too
+ if not action_completed and task_list:
+ try:
+ self._insert_pending_tasks(task_list)
+ except Exception as e:
+ self.logger.critical("Unexpected exception at _reload_vim_actions:_insert_pending_tasks: " + str(e),
+ exc_info=True)
+ self.logger.debug("reloaded vim actions pending:{} refresh:{}".format(
+ len(self.pending_tasks), len(self.refresh_tasks)))
+ except Exception as e:
+ self.logger.critical("Unexpected exception at _reload_vim_actions: " + str(e), exc_info=True)
+
+ def _refres_elements(self):
+ """Call VIM to get VMs and networks status until 10 elements"""
+ now = time.time()
+ nb_processed = 0
+ vm_to_refresh_list = []
+ net_to_refresh_list = []
+ vm_to_refresh_dict = {}
+ net_to_refresh_dict = {}
+ items_to_refresh = 0
+ while self.refresh_tasks:
+ task = self.refresh_tasks[0]
+ with self.task_lock:
+ if task['status'] == 'SUPERSEDED':
+ self.refresh_tasks.pop(0)
+ continue
+ if task['modified_at'] > now:
+ break
+ # task["status"] = "processing"
+ nb_processed += 1
+ self.refresh_tasks.pop(0)
+ if task["item"] == 'instance_vms':
+ vm_to_refresh_list.append(task["vim_id"])
+ vm_to_refresh_dict[task["vim_id"]] = task
+ elif task["item"] == 'instance_nets':
+ net_to_refresh_list.append(task["vim_id"])
+ net_to_refresh_dict[task["vim_id"]] = task
+ else:
+ task_id = task["instance_action_id"] + "." + str(task["task_index"])
+ self.logger.critical("task={}: unknown task {}".format(task_id, task["item"]), exc_info=True)
+ items_to_refresh += 1
+ if items_to_refresh == 10:
+ break
+
+ if vm_to_refresh_list:
+ now = time.time()
+ try:
+ vim_dict = self.vim.refresh_vms_status(vm_to_refresh_list)
+ except vimconn.vimconnException as e:
+ # Mark all tasks at VIM_ERROR status
+ self.logger.error("task=several get-VM: vimconnException when trying to refresh vms " + str(e))
+ vim_dict = {}
+ for vim_id in vm_to_refresh_list:
+ vim_dict[vim_id] = {"status": "VIM_ERROR", "error_msg": str(e)}
+
+ for vim_id, vim_info in vim_dict.items():
+ # look for task
+ task_need_update = False
+ task = vm_to_refresh_dict[vim_id]
+ task_id = task["instance_action_id"] + "." + str(task["task_index"])
+ self.logger.debug("task={} get-VM: vim_vm_id={} result={}".format(task_id, task["vim_id"], vim_info))
+
+ # check and update interfaces
+ task_warning_msg = ""
+ for interface in vim_info.get("interfaces", ()):
+ vim_interface_id = interface["vim_interface_id"]
+ if vim_interface_id not in task["extra"]["interfaces"]:
+ self.logger.critical("task={} get-VM: Interface not found {} on task info {}".format(
+ task_id, vim_interface_id, task["extra"]["interfaces"]), exc_info=True)
+ continue
+ task_interface = task["extra"]["interfaces"][vim_interface_id]
+ task_vim_interface = task["vim_interfaces"].get(vim_interface_id)
+ if task_vim_interface != interface:
+ # delete old port
+ if task_interface.get("sdn_port_id"):
+ try:
+ with self.db_lock:
+ self.ovim.delete_port(task_interface["sdn_port_id"])
+ task_interface["sdn_port_id"] = None
+ task_need_update = True
+ except ovimException as e:
+ error_text = "ovimException deleting external_port={}: {}".format(
+ task_interface["sdn_port_id"], e)
+ self.logger.error("task={} get-VM: {}".format(task_id, error_text), exc_info=True)
+ task_warning_msg += error_text
+ # TODO Set error_msg at instance_nets instead of instance VMs
+
+ # Create SDN port
+ sdn_net_id = task_interface.get("sdn_net_id")
+ if sdn_net_id and interface.get("compute_node") and interface.get("pci"):
+ sdn_port_name = sdn_net_id + "." + task["vim_id"]
+ sdn_port_name = sdn_port_name[:63]
+ try:
+ with self.db_lock:
+ 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")})
+ task_interface["sdn_port_id"] = sdn_port_id
+ task_need_update = True
+ except (ovimException, Exception) as e:
+ error_text = "ovimException creating new_external_port compute_node={}"\
+ " pci={} vlan={} {}".format(
+ interface["compute_node"],
+ interface["pci"],
+ interface.get("vlan"), e)
+ self.logger.error("task={} get-VM: {}".format(task_id, error_text), exc_info=True)
+ task_warning_msg += error_text
+ # TODO Set error_msg at instance_nets instead of instance VMs
+
+ with self.db_lock:
+ self.db.update_rows(
+ 'instance_interfaces',
+ UPDATE={"mac_address": interface.get("mac_address"),
+ "ip_address": interface.get("ip_address"),
+ "vim_info": interface.get("vim_info"),
+ "sdn_port_id": task_interface.get("sdn_port_id"),
+ "compute_node": interface.get("compute_node"),
+ "pci": interface.get("pci"),
+ "vlan": interface.get("vlan")},
+ WHERE={'uuid': task_interface["iface_id"]})
+ task["vim_interfaces"][vim_interface_id] = interface
+
+ # check and update task and instance_vms database
+ if vim_info.get("error_msg"):
+ vim_info["error_msg"] = self._format_vim_error_msg(vim_info["error_msg"] + task_warning_msg)
+ elif task_warning_msg:
+ vim_info["error_msg"] = self._format_vim_error_msg(task_warning_msg)
+
+ task_vim_info = task.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.get("error_msg") or \
+ (vim_info.get("vim_info") and task_vim_info != vim_info["vim_info"]):
+ temp_dict = {"status": vim_info["status"], "error_msg": vim_info.get("error_msg")}
+ if vim_info.get("vim_info"):
+ temp_dict["vim_info"] = vim_info["vim_info"]
+ with self.db_lock:
+ self.db.update_rows('instance_vms', UPDATE=temp_dict, WHERE={"uuid": task["item_id"]})
+ task["extra"]["vim_status"] = vim_info["status"]
+ task["error_msg"] = vim_info.get("error_msg")
+ if vim_info.get("vim_info"):
+ task["vim_info"] = vim_info["vim_info"]
+ task_need_update = True
+
+ if task_need_update:
+ with self.db_lock:
+ self.db.update_rows(
+ 'vim_actions',
+ UPDATE={"extra": yaml.safe_dump(task["extra"], default_flow_style=True, width=256),
+ "error_msg": task.get("error_msg"), "modified_at": now},
+ WHERE={'instance_action_id': task['instance_action_id'],
+ 'task_index': task['task_index']})
+ if task["extra"].get("vim_status") == "BUILD":
+ self._insert_refresh(task, now + self.REFRESH_BUILD)
+ else:
+ self._insert_refresh(task, now + self.REFRESH_ACTIVE)
+
+ if net_to_refresh_list:
+ now = time.time()
+ try:
+ vim_dict = self.vim.refresh_nets_status(net_to_refresh_list)
+ except vimconn.vimconnException as e:
+ # Mark all tasks at VIM_ERROR status
+ self.logger.error("task=several get-net: vimconnException when trying to refresh nets " + str(e))
+ vim_dict = {}
+ for vim_id in net_to_refresh_list:
+ vim_dict[vim_id] = {"status": "VIM_ERROR", "error_msg": str(e)}
+
+ for vim_id, vim_info in vim_dict.items():
+ # look for task
+ task = net_to_refresh_dict[vim_id]
+ task_id = task["instance_action_id"] + "." + str(task["task_index"])
+ self.logger.debug("task={} get-net: vim_net_id={} result={}".format(task_id, task["vim_id"], vim_info))
+
+ task_vim_info = task.get("vim_info")
+ task_vim_status = task["extra"].get("vim_status")
+ task_error_msg = task.get("error_msg")
+ task_sdn_net_id = task["extra"].get("sdn_net_id")
+
+ # get ovim status
+ if task_sdn_net_id:
+ try:
+ with self.db_lock:
+ sdn_net = self.ovim.show_network(task_sdn_net_id)
+ except (ovimException, Exception) as e:
+ text_error = "ovimException getting network snd_net_id={}: {}".format(task_sdn_net_id, e)
+ self.logger.error("task={} get-net: {}".format(task_id, text_error), exc_info=True)
+ sdn_net = {"status": "ERROR", "error_msg": text_error}
+ 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"
+
+ # update database
+ if vim_info.get("error_msg"):
+ vim_info["error_msg"] = self._format_vim_error_msg(vim_info["error_msg"])
+ if task_vim_status != vim_info["status"] or task_error_msg != vim_info.get("error_msg") or \
+ (vim_info.get("vim_info") and task_vim_info != vim_info["vim_info"]):
+ task["extra"]["vim_status"] = vim_info["status"]
+ task["error_msg"] = vim_info.get("error_msg")
+ if vim_info.get("vim_info"):
+ task["vim_info"] = vim_info["vim_info"]
+ temp_dict = {"status": vim_info["status"], "error_msg": vim_info.get("error_msg")}
+ if vim_info.get("vim_info"):
+ temp_dict["vim_info"] = vim_info["vim_info"]
+ with self.db_lock:
+ self.db.update_rows('instance_nets', UPDATE=temp_dict, WHERE={"uuid": task["item_id"]})
+ self.db.update_rows(
+ 'vim_actions',
+ UPDATE={"extra": yaml.safe_dump(task["extra"], default_flow_style=True, width=256),
+ "error_msg": task.get("error_msg"), "modified_at": now},
+ WHERE={'instance_action_id': task['instance_action_id'],
+ 'task_index': task['task_index']})
+ if task["extra"].get("vim_status") == "BUILD":
+ self._insert_refresh(task, now + self.REFRESH_BUILD)
+ else:
+ self._insert_refresh(task, now + self.REFRESH_ACTIVE)
+
+ return nb_processed
+
+ def _insert_refresh(self, task, threshold_time=None):
+ """Insert a task at list of refreshing elements. The refreshing list is ordered by threshold_time (task['modified_at']
+ It is assumed that this is called inside this thread
+ """
+ if not self.vim:
+ return
+ if not threshold_time:
+ threshold_time = time.time()
+ task["modified_at"] = threshold_time
+ task_name = task["item"][9:] + "-" + task["action"]
+ task_id = task["instance_action_id"] + "." + str(task["task_index"])
+ for index in range(0, len(self.refresh_tasks)):
+ if self.refresh_tasks[index]["modified_at"] > threshold_time:
+ self.refresh_tasks.insert(index, task)
+ break
+ else:
+ index = len(self.refresh_tasks)
+ self.refresh_tasks.append(task)
+ self.logger.debug("task={} new refresh name={}, modified_at={} index={}".format(
+ task_id, task_name, task["modified_at"], 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_tasks)):
+ if self.refresh_tasks[index]["name"] == task_name and self.refresh_tasks[index]["vim_id"] == vim_id:
+ index_to_delete = index
+ break
+ else:
+ return False
+ if not index_to_delete:
+ del self.refresh_tasks[index_to_delete]
+ return True
+
+ def _proccess_pending_tasks(self):
+ nb_created = 0
+ nb_processed = 0
+ while self.pending_tasks:
+ task = self.pending_tasks.pop(0)
+ nb_processed += 1
+ try:
+ # check if tasks that this depends on have been completed
+ dependency_not_completed = False
+ for task_index in task["extra"].get("depends_on", ()):
+ task_dependency = task["depends"].get("TASK-" + str(task_index))
+ if not task_dependency:
+ task_dependency = self._look_for_task(task["instance_action_id"], "TASK-" + str(task_index))
+ if not task_dependency:
+ raise VimThreadException(
+ "Cannot get depending net task trying to get depending task {}.{}".format(
+ task["instance_action_id"], task_index))
+ # task["depends"]["TASK-" + str(task_index)] = task_dependency #it references another object,so database must be look again
+ if task_dependency["status"] == "SCHEDULED":
+ dependency_not_completed = True
+ break
+ elif task_dependency["status"] == "FAILED":
+ raise VimThreadException(
+ "Cannot {} {}, (task {}.{}) because depends on failed {} {}, (task{}.{})".format(
+ task["action"], task["item"],
+ task["instance_action_id"], task["task_index"],
+ task_dependency["instance_action_id"], task_dependency["task_index"],
+ task_dependency["action"], task_dependency["item"]))
+ if dependency_not_completed:
+ # Move this task to the end.
+ task["extra"]["tries"] = task["extra"].get("tries", 0) + 1
+ if task["extra"]["tries"] <= 3:
+ self.pending_tasks.append(task)
+ continue
+ else:
+ raise VimThreadException(
+ "Cannot {} {}, (task {}.{}) because timeout waiting to complete {} {}, "
+ "(task {}.{})".format(task["action"], task["item"],
+ task["instance_action_id"], task["task_index"],
+ task_dependency["instance_action_id"], task_dependency["task_index"],
+ task_dependency["action"], task_dependency["item"]))
+
+ if task["status"] == "SUPERSEDED":
+ # not needed to do anything but update database with the new status
+ result = True
+ database_update = None
+ elif not self.vim:
+ task["status"] = "ERROR"
+ task["error_msg"] = self.error_status
+ result = False
+ database_update = {"status": "VIM_ERROR", "error_msg": task["error_msg"]}
+ elif task["item"] == 'instance_vms':
+ if task["action"] == "CREATE":
+ result, database_update = self.new_vm(task)
+ nb_created += 1
+ elif task["action"] == "DELETE":
+ result, database_update = self.del_vm(task)
+ else:
+ raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
+ elif task["item"] == 'instance_nets':
+ if task["action"] == "CREATE":
+ result, database_update = self.new_net(task)
+ nb_created += 1
+ elif task["action"] == "DELETE":
+ result, database_update = self.del_net(task)
+ elif task["action"] == "FIND":
+ result, database_update = self.get_net(task)
+ else:
+ raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
+ elif task["item"] == 'instance_sfis':
+ if task["action"] == "CREATE":
+ result, database_update = self.new_sfi(task)
+ nb_created += 1
+ elif task["action"] == "DELETE":
+ result, database_update = self.del_sfi(task)
+ else:
+ raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
+ elif task["item"] == 'instance_sfs':
+ if task["action"] == "CREATE":
+ result, database_update = self.new_sf(task)
+ nb_created += 1
+ elif task["action"] == "DELETE":
+ result, database_update = self.del_sf(task)
+ else:
+ raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
+ elif task["item"] == 'instance_classifications':
+ if task["action"] == "CREATE":
+ result, database_update = self.new_classification(task)
+ nb_created += 1
+ elif task["action"] == "DELETE":
+ result, database_update = self.del_classification(task)
+ else:
+ raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
+ elif task["item"] == 'instance_sfps':
+ if task["action"] == "CREATE":
+ result, database_update = self.new_sfp(task)
+ nb_created += 1
+ elif task["action"] == "DELETE":
+ result, database_update = self.del_sfp(task)
+ else:
+ raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
+ else:
+ raise vimconn.vimconnException(self.name + "unknown task item {}".format(task["item"]))
+ # TODO
+ except VimThreadException as e:
+ result = False
+ 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
+
+ no_refresh_tasks = ['instance_sfis', 'instance_sfs',
+ 'instance_classifications', 'instance_sfps']
+ if task["action"] == "DELETE":
+ action_key = task["item"] + task["item_id"]
+ del self.grouped_tasks[action_key]
+ elif task["action"] in ("CREATE", "FIND") and task["status"] in ("DONE", "BUILD"):
+ if task["item"] not in no_refresh_tasks:
+ self._insert_refresh(task)
+
+ 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"]))
+ try:
+ now = time.time()
+ with self.db_lock:
+ self.db.update_rows(
+ table="vim_actions",
+ UPDATE={"status": task["status"], "vim_id": task.get("vim_id"), "modified_at": now,
+ "error_msg": task["error_msg"],
+ "extra": yaml.safe_dump(task["extra"], default_flow_style=True, width=256)},
+ WHERE={"instance_action_id": task["instance_action_id"], "task_index": task["task_index"]})
+ if result is not None:
+ self.db.update_rows(
+ table="instance_actions",
+ UPDATE={("number_done" if result else "number_failed"): {"INCREMENT": 1},
+ "modified_at": now},
+ WHERE={"uuid": task["instance_action_id"]})
+ if database_update:
+ self.db.update_rows(table=task["item"],
+ UPDATE=database_update,
+ WHERE={"uuid": task["item_id"]})
+ except db_base_Exception as e:
+ self.logger.error("task={} Error updating database {}".format(task_id, e), exc_info=True)
+
+ if nb_created == 10:
+ break
+ return nb_processed
+
+ def _insert_pending_tasks(self, vim_actions_list):
+ for task in vim_actions_list:
+ if task["datacenter_vim_id"] != self.datacenter_tenant_id:
+ continue
+ item = task["item"]
+ item_id = task["item_id"]
+ action_key = item + item_id
+ if action_key not in self.grouped_tasks:
+ self.grouped_tasks[action_key] = []
+ task["params"] = None
+ task["depends"] = {}
+ if task["extra"]:
+ extra = yaml.load(task["extra"])
+ task["extra"] = extra
+ task["params"] = extra.get("params")
+ depends_on_list = extra.get("depends_on")
+ if depends_on_list:
+ for index in depends_on_list:
+ if index < len(vim_actions_list) and vim_actions_list[index]["task_index"] == index and\
+ vim_actions_list[index]["instance_action_id"] == task["instance_action_id"]:
+ task["depends"]["TASK-" + str(index)] = vim_actions_list[index]
+ if extra.get("interfaces"):
+ task["vim_interfaces"] = {}
+ else:
+ task["extra"] = {}
+ if "error_msg" not in task:
+ task["error_msg"] = None
+ if "vim_id" not in task:
+ task["vim_id"] = None
+
+ if task["action"] == "DELETE":
+ need_delete_action = False
+ for to_supersede in self.grouped_tasks.get(action_key, ()):
+ if to_supersede["action"] == "FIND" and to_supersede.get("vim_id"):
+ task["vim_id"] = to_supersede["vim_id"]
+ if to_supersede["action"] == "CREATE" and to_supersede["extra"].get("created", True) and \
+ (to_supersede.get("vim_id") or to_supersede["extra"].get("sdn_net_id")):
+ need_delete_action = True
+ task["vim_id"] = to_supersede["vim_id"]
+ if to_supersede["extra"].get("sdn_net_id"):
+ task["extra"]["sdn_net_id"] = to_supersede["extra"]["sdn_net_id"]
+ if to_supersede["extra"].get("interfaces"):
+ task["extra"]["interfaces"] = to_supersede["extra"]["interfaces"]
+ if to_supersede["extra"].get("created_items"):
+ if not task["extra"].get("created_items"):
+ task["extra"]["created_items"] = {}
+ task["extra"]["created_items"].update(to_supersede["extra"]["created_items"])
+ # Mark task as SUPERSEDED.
+ # If task is in self.pending_tasks, it will be removed and database will be update
+ # If task is in self.refresh_tasks, it will be removed
+ to_supersede["status"] = "SUPERSEDED"
+ if not need_delete_action:
+ task["status"] = "SUPERSEDED"
+
+ self.grouped_tasks[action_key].append(task)
+ self.pending_tasks.append(task)
+ elif task["status"] == "SCHEDULED":
+ self.grouped_tasks[action_key].append(task)
+ self.pending_tasks.append(task)
+ elif task["action"] in ("CREATE", "FIND"):
+ self.grouped_tasks[action_key].append(task)
+ if task["status"] in ("DONE", "BUILD"):
+ self._insert_refresh(task)
+ # TODO add VM reset, get console, etc...
+ else:
+ raise vimconn.vimconnException(self.name + "unknown vim_action action {}".format(task["action"]))
+