This is thread that interact with the host and the libvirt to manage VM
One thread will be launched per host
'''
-__author__ = "Alfonso Tierno"
+__author__ = "Alfonso Tierno, Pablo Montes"
__date__ = "$10-feb-2017 12:07:15$"
import threading
import logging
import vimconn
from db_base import db_base_Exception
+from ovim import ovimException
# from logging import Logger
class vim_thread(threading.Thread):
- def __init__(self, vimconn, task_lock, name=None, datacenter_name=None, datacenter_tenant_id=None, db=None, db_lock=None):
+ def __init__(self, vimconn, task_lock, name=None, datacenter_name=None, datacenter_tenant_id=None, db=None, db_lock=None, ovim=None):
"""Init a thread.
Arguments:
'id' number of thead
self.vim = vimconn
self.datacenter_name = datacenter_name
self.datacenter_tenant_id = datacenter_tenant_id
+ self.ovim = ovim
if not name:
self.name = vimconn["id"] + "." + vimconn["config"]["datacenter_tenant_id"]
else:
self.task_lock = task_lock
self.task_queue = Queue.Queue(2000)
+ self.refresh_list = []
+ """Contains time ordered task list for refreshing the status of VIM VMs and nets"""
+
+ def _refres_elements(self):
+ """Call VIM to get VMs and networks status until 10 elements"""
+ now = time.time()
+ vm_to_refresh_list = []
+ net_to_refresh_list = []
+ vm_to_refresh_dict = {}
+ net_to_refresh_dict = {}
+ items_to_refresh = 0
+ while self.refresh_list:
+ task = self.refresh_list[0]
+ with self.task_lock:
+ if task['status'] == 'deleted':
+ self.refresh_list.pop(0)
+ continue
+ if task['time'] > now:
+ break
+ task["status"] = "processing"
+ self.refresh_list.pop(0)
+ if task["name"] == 'get-vm':
+ vm_to_refresh_list.append(task["vim_id"])
+ vm_to_refresh_dict[task["vim_id"]] = task
+ elif task["name"] == 'get-net':
+ net_to_refresh_list.append(task["vim_id"])
+ net_to_refresh_dict[task["vim_id"]] = task
+ else:
+ error_text = "unknown task {}".format(task["name"])
+ self.logger.error(error_text)
+ items_to_refresh += 1
+ if items_to_refresh == 10:
+ break
+
+ if vm_to_refresh_list:
+ try:
+ vim_dict = self.vim.refresh_vms_status(vm_to_refresh_list)
+ for vim_id, vim_info in vim_dict.items():
+ #look for task
+ task = vm_to_refresh_dict[vim_id]
+ self.logger.debug("get-vm vm_id=%s result=%s", task["vim_id"], str(vim_info))
+
+ # update database
+ if vim_info.get("error_msg"):
+ vim_info["error_msg"] = self._format_vim_error_msg(vim_info["error_msg"])
+ if task["vim_info"].get("status") != vim_info["status"] or \
+ task["vim_info"].get("error_msg") != vim_info.get("error_msg") or \
+ task["vim_info"].get("vim_info") != vim_info["vim_info"]:
+ with self.db_lock:
+ temp_dict = {"status": vim_info["status"],
+ "error_msg": vim_info.get("error_msg"),
+ "vim_info": vim_info["vim_info"]}
+ self.db.update_rows('instance_vms', UPDATE=temp_dict, WHERE={"vim_vm_id": vim_id})
+ for interface in vim_info["interfaces"]:
+ for task_interface in task["vim_info"]["interfaces"]:
+ if task_interface["vim_net_id"] == interface["vim_net_id"]:
+ break
+ else:
+ task_interface = {"vim_net_id": interface["vim_net_id"]}
+ task["vim_info"]["interfaces"].append(task_interface)
+ if task_interface != interface:
+ #delete old port
+ if task_interface.get("sdn_port_id"):
+ try:
+ self.ovim.delete_port(task_interface["sdn_port_id"])
+ task_interface["sdn_port_id"] = None
+ except ovimException as e:
+ self.logger.error("ovimException deleting external_port={} ".format(
+ task_interface["sdn_port_id"]) + str(e), exc_info=True)
+ # TODO Set error_msg at instance_nets
+ vim_net_id = interface.pop("vim_net_id")
+ sdn_net_id = None
+ sdn_port_name = None
+ with self.db_lock:
+ where_= {'iv.vim_vm_id': vim_id, "ine.vim_net_id": vim_net_id,
+ 'ine.datacenter_tenant_id': self.datacenter_tenant_id}
+ # TODO check why vim_interface_id is not present at database
+ # if interface.get("vim_interface_id"):
+ # where_["vim_interface_id"] = interface["vim_interface_id"]
+ db_ifaces = self.db.get_rows(
+ FROM="instance_interfaces as ii left join instance_nets as ine on "
+ "ii.instance_net_id=ine.uuid left join instance_vms as iv on "
+ "ii.instance_vm_id=iv.uuid",
+ SELECT=("ii.uuid as iface_id", "ine.uuid as net_id", "iv.uuid as vm_id", "sdn_net_id"),
+ WHERE=where_)
+ if len(db_ifaces)>1:
+ self.logger.critical("Refresing interfaces. "
+ "Found more than one interface at database for '{}'".format(where_))
+ elif len(db_ifaces)==0:
+ self.logger.critical("Refresing interfaces. "
+ "Not found any interface at database for '{}'".format(where_))
+ continue
+ else:
+ db_iface = db_ifaces[0]
+ if db_iface.get("sdn_net_id") and interface.get("compute_node") and interface.get("pci"):
+ sdn_net_id = db_iface["sdn_net_id"]
+ sdn_port_name = sdn_net_id + "." + db_iface["vm_id"]
+ sdn_port_name = sdn_port_name[:63]
+ try:
+ 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")})
+ interface["sdn_port_id"] = sdn_port_id
+ except (ovimException, Exception) as e:
+ self.logger.error(
+ "ovimException creating new_external_port compute_node={} " \
+ "pci={} vlan={} ".format(
+ interface["compute_node"],
+ interface["pci"],
+ interface.get("vlan")) + str(e),
+ exc_info=True)
+ # TODO Set error_msg at instance_nets
+ with self.db_lock:
+ self.db.update_rows('instance_interfaces', UPDATE=interface,
+ WHERE={'uuid': db_iface["iface_id"]})
+ # TODO insert instance_id
+ interface["vim_net_id"] = vim_net_id
+
+ task["vim_info"] = vim_info
+ if task["vim_info"]["status"] == "BUILD":
+ self._insert_refresh(task, now+5) # 5seconds
+ else:
+ self._insert_refresh(task, now+300) # 5minutes
+ except vimconn.vimconnException as e:
+ self.logger.error("vimconnException Exception when trying to refresh vms " + str(e))
+ self._insert_refresh(task, now + 300) # 5minutes
+
+ if net_to_refresh_list:
+ try:
+ vim_dict = self.vim.refresh_nets_status(net_to_refresh_list)
+ for vim_id, vim_info in vim_dict.items():
+ #look for task
+ task = net_to_refresh_dict[vim_id]
+ self.logger.debug("get-net net_id=%s result=%s", task["vim_id"], str(vim_info))
+
+ #get database info
+ where_ = {"vim_net_id": vim_id, 'datacenter_tenant_id': self.datacenter_tenant_id}
+ with self.db_lock:
+ db_nets = self.db.get_rows(
+ FROM="instance_nets",
+ SELECT=("uuid as net_id", "sdn_net_id"),
+ WHERE=where_)
+ if len(db_nets) > 1:
+ self.logger.critical("Refresing networks. "
+ "Found more than one instance-networks at database for '{}'".format(where_))
+ elif len(db_nets) == 0:
+ self.logger.critical("Refresing networks. "
+ "Not found any instance-network at database for '{}'".format(where_))
+ continue
+ else:
+ db_net = db_nets[0]
+ if db_net.get("sdn_net_id"):
+ # get ovim status
+ try:
+ sdn_net = self.ovim.show_network(db_net["sdn_net_id"])
+ 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"
+
+ except (ovimException, Exception) as e:
+ self.logger.error(
+ "ovimException getting network infor snd_net_id={}".format(db_net["sdn_net_id"]),
+ exc_info=True)
+ # TODO Set error_msg at instance_nets
+
+ # update database
+ if vim_info.get("error_msg"):
+ vim_info["error_msg"] = self._format_vim_error_msg(vim_info["error_msg"])
+ if task["vim_info"].get("status") != vim_info["status"] or \
+ task["vim_info"].get("error_msg") != vim_info.get("error_msg") or \
+ task["vim_info"].get("vim_info") != vim_info["vim_info"]:
+ with self.db_lock:
+ temp_dict = {"status": vim_info["status"],
+ "error_msg": vim_info.get("error_msg"),
+ "vim_info": vim_info["vim_info"]}
+ self.db.update_rows('instance_nets', UPDATE=temp_dict, WHERE={"vim_net_id": vim_id})
+
+ task["vim_info"] = vim_info
+ if task["vim_info"]["status"] == "BUILD":
+ self._insert_refresh(task, now+5) # 5seconds
+ else:
+ self._insert_refresh(task, now+300) # 5minutes
+ except vimconn.vimconnException as e:
+ self.logger.error("vimconnException Exception when trying to refresh nets " + str(e))
+ self._insert_refresh(task, now + 300) # 5minutes
+
+ if not items_to_refresh:
+ time.sleep(1)
+
+ def _insert_refresh(self, task, threshold_time):
+ """Insert a task at list of refreshing elements. The refreshing list is ordered by threshold_time (task['time']
+ It is assumed that this is called inside this thread
+ """
+ task["time"] = threshold_time
+ for index in range(0, len(self.refresh_list)):
+ if self.refresh_list[index]["time"] > threshold_time:
+ self.refresh_list.insert(index, task)
+ break
+ else:
+ index = len(self.refresh_list)
+ self.refresh_list.append(task)
+ self.logger.debug("new refresh task={} name={}, time={} index={}".format(
+ task["id"], task["name"], task["time"], 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_list)):
+ if self.refresh_list[index]["name"] == task_name and self.refresh_list[index]["vim_id"] == vim_id:
+ index_to_delete = index
+ break
+ else:
+ return False
+ if index_to_delete != None:
+ del self.refresh_list[index_to_delete]
+ return True
def insert_task(self, task):
try:
while True:
#TODO reload service
while True:
- if not self.task_queue.empty():
- task = self.task_queue.get()
- self.task_lock.acquire()
- if task["status"] == "deleted":
+ try:
+ if not self.task_queue.empty():
+ task = self.task_queue.get()
+ self.task_lock.acquire()
+ if task["status"] == "deleted":
+ self.task_lock.release()
+ continue
+ task["status"] = "processing"
self.task_lock.release()
+ else:
+ self._refres_elements()
continue
- task["status"] == "processing"
- self.task_lock.release()
- else:
- now=time.time()
- time.sleep(1)
- continue
- self.logger.debug("processing task id={} name={} params={}".format(task["id"], task["name"],
- str(task["params"])))
- if task["name"] == 'exit' or task["name"] == 'reload':
- result, content = self.terminate(task)
- elif task["name"] == 'new-vm':
- result, content = self.new_vm(task)
- elif task["name"] == 'del-vm':
- result, content = self.del_vm(task)
- elif task["name"] == 'new-net':
- result, content = self.new_net(task)
- elif task["name"] == 'del-net':
- result, content = self.del_net(task)
- else:
- error_text = "unknown task {}".format(task["name"])
- self.logger.error(error_text)
- result = False
- content = error_text
+ self.logger.debug("processing task id={} name={} params={}".format(task["id"], task["name"],
+ str(task["params"])))
+ if task["name"] == 'exit' or task["name"] == 'reload':
+ result, content = self.terminate(task)
+ elif task["name"] == 'new-vm':
+ result, content = self.new_vm(task)
+ elif task["name"] == 'del-vm':
+ result, content = self.del_vm(task)
+ elif task["name"] == 'new-net':
+ result, content = self.new_net(task)
+ elif task["name"] == 'del-net':
+ result, content = self.del_net(task)
+ else:
+ error_text = "unknown task {}".format(task["name"])
+ self.logger.error(error_text)
+ result = False
+ content = error_text
+ self.logger.debug("task id={} name={} result={}:{} params={}".format(task["id"], task["name"],
+ result, content,
+ str(task["params"])))
- with self.task_lock:
- task["status"] = "done" if result else "error"
- task["result"] = content
- self.task_queue.task_done()
+ with self.task_lock:
+ task["status"] = "done" if result else "error"
+ task["result"] = content
+ self.task_queue.task_done()
- if task["name"] == 'exit':
- return 0
- elif task["name"] == 'reload':
- break
+ if task["name"] == 'exit':
+ return 0
+ elif task["name"] == 'reload':
+ break
+ except Exception as e:
+ self.logger.critical("Unexpected exception at run: " + str(e), exc_info=True)
self.logger.debug("Finishing")
def terminate(self, task):
return True, None
+ def _format_vim_error_msg(self, error_text, max_length=1024):
+ if error_text and len(error_text) >= max_length:
+ return error_text[:max_length//2-3] + " ... " + error_text[-max_length//2+3:]
+ return error_text
+
def new_net(self, task):
try:
task_id = task["id"]
params = task["params"]
net_id = self.vim.new_network(*params)
+
+ net_name = params[0]
+ net_type = params[1]
+
+ network = None
+ sdn_net_id = None
+ sdn_controller = self.vim.config.get('sdn-controller')
+ if sdn_controller and (net_type == "data" or net_type == "ptp"):
+ network = {"name": net_name, "type": net_type}
+
+ vim_net = self.vim.get_network(net_id)
+ if vim_net.get('encapsulation') != 'vlan':
+ raise vimconn.vimconnException(
+ "net '{}' defined as type '{}' has not vlan encapsulation '{}'".format(
+ net_name, net_type, vim_net['encapsulation']))
+ network["vlan"] = vim_net.get('segmentation_id')
+ try:
+ sdn_net_id = self.ovim.new_network(network)
+ except (ovimException, Exception) as e:
+ self.logger.error("task=%s cannot create SDN network vim_net_id=%s input='%s' ovimException='%s'",
+ str(task_id), net_id, str(network), str(e))
with self.db_lock:
- self.db.update_rows("instance_nets", UPDATE={"vim_net_id": net_id}, WHERE={"vim_net_id": task_id})
+ self.db.update_rows("instance_nets", UPDATE={"vim_net_id": net_id, "sdn_net_id": sdn_net_id},
+ WHERE={"vim_net_id": task_id})
+ new_refresh_task = {"status": "enqueued",
+ "id": task_id,
+ "name": "get-net",
+ "vim_id": net_id,
+ "vim_info": {} }
+ self._insert_refresh(new_refresh_task, time.time())
return True, net_id
except db_base_Exception as e:
self.logger.error("Error updating database %s", str(e))
return True, net_id
except vimconn.vimconnException as e:
+ self.logger.error("Error creating NET, task=%s: %s", str(task_id), str(e))
+ try:
+ with self.db_lock:
+ self.db.update_rows("instance_nets",
+ UPDATE={"error_msg": self._format_vim_error_msg(str(e)), "status": "VIM_ERROR"},
+ WHERE={"vim_net_id": task_id})
+ except db_base_Exception as e:
+ self.logger.error("Error updating database %s", str(e))
return False, str(e)
+ #except ovimException as e:
+ # self.logger.error("Error creating NET in ovim, task=%s: %s", str(task_id), str(e))
+ # return False, str(e)
def new_vm(self, task):
try:
task_id = task["id"]
depends = task.get("depends")
net_list = params[5]
+ error_text = ""
for net in net_list:
- if is_task_id(net["net_id"]): # change task_id into network_id
+ if "net_id" in net and is_task_id(net["net_id"]): # change task_id into network_id
try:
task_net = depends[net["net_id"]]
with self.task_lock:
if task_net["status"] == "error":
- return False, "Cannot create VM because depends on a network that cannot be created: " + \
+ error_text = "Cannot create VM because depends on a network that cannot be created: " +\
str(task_net["result"])
+ break
elif task_net["status"] == "enqueued" or task_net["status"] == "processing":
- return False, "Cannot create VM because depends on a network still not created"
+ error_text = "Cannot create VM because depends on a network still not created"
+ break
network_id = task_net["result"]
net["net_id"] = network_id
except Exception as e:
- return False, "Error trying to map from task_id={} to task result: {}".format(net["net_id"],
- str(e))
- vm_id = self.vim.new_vminstance(*params)
- with self.db_lock:
- self.db.update_rows("instance_vms", UPDATE={"vim_vm_id": vm_id}, WHERE={"vim_vm_id": task_id})
- return True, vm_id
- except db_base_Exception as e:
- self.logger.error("Error updtaing database %s", str(e))
+ error_text = "Error trying to map from task_id={} to task result: {}".format(
+ net["net_id"],str(e))
+ break
+ if not error_text:
+ vm_id = self.vim.new_vminstance(*params)
+ try:
+ with self.db_lock:
+ if error_text:
+ update = self.db.update_rows("instance_vms",
+ UPDATE={"status": "VIM_ERROR", "error_msg": error_text},
+ WHERE={"vim_vm_id": task_id})
+ else:
+ update = self.db.update_rows("instance_vms", UPDATE={"vim_vm_id": vm_id}, WHERE={"vim_vm_id": task_id})
+ if not update:
+ self.logger.error("task id={} name={} database not updated vim_vm_id={}".format(
+ task["id"], task["name"], vm_id))
+ except db_base_Exception as e:
+ self.logger.error("Error updating database %s", str(e))
+ if error_text:
+ return False, error_text
+ new_refresh_task = {"status": "enqueued",
+ "id": task_id,
+ "name": "get-vm",
+ "vim_id": vm_id,
+ "vim_info": {"interfaces":[]} }
+ self._insert_refresh(new_refresh_task, time.time())
return True, vm_id
except vimconn.vimconnException as e:
+ self.logger.error("Error creating VM, task=%s: %s", str(task_id), str(e))
+ try:
+ with self.db_lock:
+ self.db.update_rows("instance_vms",
+ UPDATE={"error_msg": self._format_vim_error_msg(str(e)), "status": "VIM_ERROR"},
+ WHERE={"vim_vm_id": task_id})
+ except db_base_Exception as edb:
+ self.logger.error("Error updating database %s", str(edb))
return False, str(e)
def del_vm(self, task):
- vm_id = task["params"]
+ vm_id = task["params"][0]
+ interfaces = task["params"][1]
if is_task_id(vm_id):
try:
task_create = task["depends"][vm_id]
if task_create["status"] == "error":
return True, "VM was not created. It has error: " + str(task_create["result"])
elif task_create["status"] == "enqueued" or task_create["status"] == "processing":
- return False, "Cannot delete VM because still creating"
+ return False, "Cannot delete VM vim_id={} because still creating".format(vm_id)
vm_id = task_create["result"]
except Exception as e:
return False, "Error trying to get task_id='{}':".format(vm_id, str(e))
try:
+ self._remove_refresh("get-vm", vm_id)
+ for iface in interfaces:
+ if iface.get("sdn_port_id"):
+ try:
+ self.ovim.delete_port(iface["sdn_port_id"])
+ except ovimException as e:
+ self.logger.error("ovimException deleting external_port={} at VM vim_id={} deletion ".format(
+ iface["sdn_port_id"], vm_id) + str(e), exc_info=True)
+ # TODO Set error_msg at instance_nets
+
return True, self.vim.delete_vminstance(vm_id)
except vimconn.vimconnException as e:
return False, str(e)
def del_net(self, task):
- net_id = task["params"]
+ net_id = task["params"][0]
+ sdn_net_id = task["params"][1]
if is_task_id(net_id):
try:
task_create = task["depends"][net_id]
except Exception as e:
return False, "Error trying to get task_id='{}':".format(net_id, str(e))
try:
- return True, self.vim.delete_network(net_id)
+ self._remove_refresh("get-net", net_id)
+ result = self.vim.delete_network(net_id)
+ if sdn_net_id:
+ with self.db_lock:
+ self.ovim.delete_network(sdn_net_id)
+ return True, result
except vimconn.vimconnException as e:
return False, str(e)
+ except ovimException as e:
+ logging.error("Error deleting network from ovim. net_id: {}, sdn_net_id: {}".format(net_id, sdn_net_id))
+ return False, str(e)
projects / osm / RO.git / blobdiff