vim_thread.py

   1 # -*- coding: utf-8 -*-
   2
   3 ##
   4 # Copyright 2015 Telefónica Investigación y Desarrollo, S.A.U.
   5 # This file is part of openvim
   6 # All Rights Reserved.
   7 #
   8 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   9 # not use this file except in compliance with the License. You may obtain
  10 # a copy of the License at
  11 #
  12 #         http://www.apache.org/licenses/LICENSE-2.0
  13 #
  14 # Unless required by applicable law or agreed to in writing, software
  15 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  16 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  17 # License for the specific language governing permissions and limitations
  18 # under the License.
  19 #
  20 # For those usages not covered by the Apache License, Version 2.0 please
  21 # contact with: nfvlabs@tid.es
  22 ##
  23
  24 '''
  25 This is thread that interact with the host and the libvirt to manage VM
  26 One thread will be launched per host
  27 '''
  28 __author__ = "Alfonso Tierno, Pablo Montes"
  29 __date__ = "$10-feb-2017 12:07:15$"
  30
  31 import threading
  32 import time
  33 import Queue
  34 import logging
  35 import vimconn
  36 from db_base import db_base_Exception
  37 #from openvim.ovim import ovimException
  38
  39
  40 # from logging import Logger
  41 # import auxiliary_functions as af
  42
  43
  44 def is_task_id(id):
  45     return True if id[:5] == "TASK." else False
  46
  47
  48 class vim_thread(threading.Thread):
  49
  50     def __init__(self, vimconn, task_lock, name=None, datacenter_name=None, datacenter_tenant_id=None, db=None, db_lock=None, ovim=None):
  51         """Init a thread.
  52         Arguments:
  53             'id' number of thead
  54             'name' name of thread
  55             'host','user':  host ip or name to manage and user
  56             'db', 'db_lock': database class and lock to use it in exclusion
  57         """
  58         self.tasksResult = {}
  59         """ It will contain a dictionary with
  60             task_id:
  61                 status: enqueued,done,error,deleted,processing
  62                 result: VIM result,
  63         """
  64         threading.Thread.__init__(self)
  65         self.vim = vimconn
  66         self.datacenter_name = datacenter_name
  67         self.datacenter_tenant_id = datacenter_tenant_id
  68         self.ovim = ovim
  69         if not name:
  70             self.name = vimconn["id"] + "." + vimconn["config"]["datacenter_tenant_id"]
  71         else:
  72             self.name = name
  73
  74         self.logger = logging.getLogger('openmano.vim.'+self.name)
  75         self.db = db
  76         self.db_lock = db_lock
  77
  78         self.task_lock = task_lock
  79         self.task_queue = Queue.Queue(2000)
  80
  81     def insert_task(self, task):
  82         try:
  83             self.task_queue.put(task, False)
  84             return task["id"]
  85         except Queue.Full:
  86             raise vimconn.vimconnException(self.name + ": timeout inserting a task")
  87
  88     def del_task(self, task):
  89         with self.task_lock:
  90             if task["status"] == "enqueued":
  91                 task["status"] == "deleted"
  92                 return True
  93             else:   # task["status"] == "processing"
  94                 self.task_lock.release()
  95                 return False
  96
  97     def run(self):
  98         self.logger.debug("Starting")
  99         while True:
 100             #TODO reload service
 101             while True:
 102                 if not self.task_queue.empty():
 103                     task = self.task_queue.get()
 104                     self.task_lock.acquire()
 105                     if task["status"] == "deleted":
 106                         self.task_lock.release()
 107                         continue
 108                     task["status"] == "processing"
 109                     self.task_lock.release()
 110                 else:
 111                     now=time.time()
 112                     time.sleep(1)
 113                     continue
 114                 self.logger.debug("processing task id={} name={} params={}".format(task["id"], task["name"],
 115                                                                                    str(task["params"])))
 116                 if task["name"] == 'exit' or task["name"] == 'reload':
 117                     result, content = self.terminate(task)
 118                 elif task["name"] == 'new-vm':
 119                     result, content = self.new_vm(task)
 120                 elif task["name"] == 'del-vm':
 121                     result, content = self.del_vm(task)
 122                 elif task["name"] == 'new-net':
 123                     result, content = self.new_net(task)
 124                 elif task["name"] == 'del-net':
 125                     result, content = self.del_net(task)
 126                 else:
 127                     error_text = "unknown task {}".format(task["name"])
 128                     self.logger.error(error_text)
 129                     result = False
 130                     content = error_text
 131
 132                 with self.task_lock:
 133                     task["status"] = "done" if result else "error"
 134                     task["result"] = content
 135                 self.task_queue.task_done()
 136
 137                 if task["name"] == 'exit':
 138                     return 0
 139                 elif task["name"] == 'reload':
 140                     break
 141
 142         self.logger.debug("Finishing")
 143
 144     def terminate(self, task):
 145         return True, None
 146
 147     def _format_vim_error_msg(self, error_text):
 148         if len(error_text) >= 1024:
 149             return error_text[:516] + " ... " + error_text[-500:]
 150         return error_text
 151
 152     def new_net(self, task):
 153         try:
 154             task_id = task["id"]
 155             params = task["params"]
 156             net_id = self.vim.new_network(*params)
 157
 158             net_name = params[0]
 159             net_type = params[1]
 160
 161             network = None
 162             #sdn_controller = self.vim.config.get('sdn-controller')
 163             sdn_controller = None
 164             if sdn_controller and (net_type == "data" or net_type == "ptp"):
 165                 network = {"name": net_name, "type": net_type}
 166
 167                 vim_net = self.vim.get_network(net_id)
 168                 if vim_net.get('encapsulation') != 'vlan':
 169                     raise vimconn.vimconnException(net_name + "defined as type " + net_type + " but the created network in vim is " + vim_net['encapsulation'])
 170
 171                 network["vlan"] = vim_net.get('segmentation_id')
 172
 173             sdn_net_id = None
 174             with self.db_lock:
 175                 if network:
 176                     sdn_net_id = self.ovim.new_network(network)
 177                 self.db.update_rows("instance_nets", UPDATE={"vim_net_id": net_id, "sdn_net_id": sdn_net_id}, WHERE={"vim_net_id": task_id})
 178
 179             return True, net_id
 180         except db_base_Exception as e:
 181             self.logger.error("Error updating database %s", str(e))
 182             return True, net_id
 183         except vimconn.vimconnException as e:
 184             self.logger.error("Error creating NET, task=%s: %s", str(task_id), str(e))
 185             try:
 186                 with self.db_lock:
 187                     self.db.update_rows("instance_nets",
 188                                         UPDATE={"error_msg": self._format_vim_error_msg(str(e)), "status": "VIM_ERROR"},
 189                                         WHERE={"vim_net_id": task_id})
 190             except db_base_Exception as e:
 191                 self.logger.error("Error updating database %s", str(e))
 192             return False, str(e)
 193         # except ovimException as e:
 194         #     self.logger.error("Error creating NET in ovim, task=%s: %s", str(task_id), str(e))
 195         #     return False, str(e)
 196
 197     def new_vm(self, task):
 198         try:
 199             params = task["params"]
 200             task_id = task["id"]
 201             depends = task.get("depends")
 202             net_list = params[5]
 203             for net in net_list:
 204                 if is_task_id(net["net_id"]):  # change task_id into network_id
 205                     try:
 206                         task_net = depends[net["net_id"]]
 207                         with self.task_lock:
 208                             if task_net["status"] == "error":
 209                                 return False, "Cannot create VM because depends on a network that cannot be created: " + \
 210                                        str(task_net["result"])
 211                             elif task_net["status"] == "enqueued" or task_net["status"] == "processing":
 212                                 return False, "Cannot create VM because depends on a network still not created"
 213                             network_id = task_net["result"]
 214                         net["net_id"] = network_id
 215                     except Exception as e:
 216                         return False, "Error trying to map from task_id={} to task result: {}".format(net["net_id"],
 217                                                                                                       str(e))
 218             vm_id = self.vim.new_vminstance(*params)
 219             try:
 220                 with self.db_lock:
 221                     self.db.update_rows("instance_vms", UPDATE={"vim_vm_id": vm_id}, WHERE={"vim_vm_id": task_id})
 222             except db_base_Exception as e:
 223                 self.logger.error("Error updating database %s", str(e))
 224             return True, vm_id
 225         except vimconn.vimconnException as e:
 226             self.logger.error("Error creating VM, task=%s: %s", str(task_id), str(e))
 227             try:
 228                 with self.db_lock:
 229                     self.db.update_rows("instance_vms",
 230                                         UPDATE={"error_msg": self._format_vim_error_msg(str(e)), "status": "VIM_ERROR"},
 231                                         WHERE={"vim_vm_id": task_id})
 232             except db_base_Exception as e:
 233                 self.logger.error("Error updating database %s", str(e))
 234             return False, str(e)
 235
 236     def del_vm(self, task):
 237         vm_id = task["params"]
 238         if is_task_id(vm_id):
 239             try:
 240                 task_create = task["depends"][vm_id]
 241                 with self.task_lock:
 242                     if task_create["status"] == "error":
 243                         return True, "VM was not created. It has error: " + str(task_create["result"])
 244                     elif task_create["status"] == "enqueued" or task_create["status"] == "processing":
 245                         return False, "Cannot delete VM because still creating"
 246                     vm_id = task_create["result"]
 247             except Exception as e:
 248                 return False, "Error trying to get task_id='{}':".format(vm_id, str(e))
 249         try:
 250             return True, self.vim.delete_vminstance(vm_id)
 251         except vimconn.vimconnException as e:
 252             return False, str(e)
 253
 254     def del_net(self, task):
 255         net_id = task["params"][0]
 256         #sdn_net_id = task["params"][1]
 257         sdn_net_id = None
 258         if is_task_id(net_id):
 259             try:
 260                 task_create = task["depends"][net_id]
 261                 with self.task_lock:
 262                     if task_create["status"] == "error":
 263                         return True, "net was not created. It has error: " + str(task_create["result"])
 264                     elif task_create["status"] == "enqueued" or task_create["status"] == "processing":
 265                         return False, "Cannot delete net because still creating"
 266                     net_id = task_create["result"]
 267             except Exception as e:
 268                 return False, "Error trying to get task_id='{}':".format(net_id, str(e))
 269         try:
 270             result = self.vim.delete_network(net_id)
 271             if sdn_net_id:
 272                 with self.db_lock:
 273                     self.ovim.delete_network(sdn_net_id)
 274             return True, result
 275         except vimconn.vimconnException as e:
 276             return False, str(e)
 277         # except ovimException as e:
 278         #     logging.error("Error deleting network from ovim. net_id: {}, sdn_net_id: {}".format(net_id, sdn_net_id))
 279         #     return False, str(e)
 280