NG-RO/osm_ng_ro/ns_thread.py

   1 # -*- coding: utf-8 -*-
   2
   3 ##
   4 # Copyright 2020 Telefonica Investigacion y Desarrollo, S.A.U.
   5 #
   6 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   7 # not use this file except in compliance with the License. You may obtain
   8 # a copy of the License at
   9 #
  10 #         http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 # Unless required by applicable law or agreed to in writing, software
  13 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  14 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  15 # License for the specific language governing permissions and limitations
  16 # under the License.
  17 #
  18 ##
  19
  20 """"
  21 This is thread that interacts with a VIM. It processes TASKs sequentially against a single VIM.
  22 The tasks are stored at database in table ro_tasks
  23 A single ro_task refers to a VIM element (flavor, image, network, ...).
  24 A ro_task can contain several 'tasks', each one with a target, where to store the results
  25 """
  26
  27 import threading
  28 import time
  29 import queue
  30 import logging
  31 import yaml
  32 from pkg_resources import iter_entry_points
  33 # from osm_common import dbmongo, dbmemory, fslocal, fsmongo, msglocal, msgkafka, version as common_version
  34 from osm_common.dbbase import DbException
  35 from osm_ro_plugin.vim_dummy import VimDummyConnector
  36 from osm_ro_plugin.sdn_dummy import SdnDummyConnector
  37 from osm_ro_plugin import vimconn, sdnconn
  38 from osm_ng_ro.vim_admin import LockRenew
  39 from copy import deepcopy
  40 from unittest.mock import Mock
  41 from http import HTTPStatus
  42 from os import mkdir
  43 from shutil import rmtree
  44
  45 __author__ = "Alfonso Tierno"
  46 __date__ = "$28-Sep-2017 12:07:15$"
  47
  48
  49 def deep_get(target_dict, *args, **kwargs):
  50     """
  51     Get a value from target_dict entering in the nested keys. If keys does not exist, it returns None
  52     Example target_dict={a: {b: 5}}; key_list=[a,b] returns 5; both key_list=[a,b,c] and key_list=[f,h] return None
  53     :param target_dict: dictionary to be read
  54     :param args: list of keys to read from  target_dict
  55     :param kwargs: only can contain default=value to return if key is not present in the nested dictionary
  56     :return: The wanted value if exist, None or default otherwise
  57     """
  58     for key in args:
  59         if not isinstance(target_dict, dict) or key not in target_dict:
  60             return kwargs.get("default")
  61         target_dict = target_dict[key]
  62     return target_dict
  63
  64
  65 class NsWorkerException(Exception):
  66     pass
  67
  68
  69 class FailingConnector:
  70     def __init__(self, error_msg):
  71         self.error_msg = error_msg
  72         for method in dir(vimconn.VimConnector):
  73             if method[0] != "_":
  74                 setattr(self, method, Mock(side_effect=vimconn.VimConnException(error_msg)))
  75         for method in dir(sdnconn.SdnConnectorBase):
  76             if method[0] != "_":
  77                 setattr(self, method, Mock(side_effect=sdnconn.SdnConnectorError(error_msg)))
  78
  79
  80 class NsWorkerExceptionNotFound(NsWorkerException):
  81     pass
  82
  83
  84 class VimInteractionBase:
  85     """ Base class to call VIM/SDN for creating, deleting and refresh networks, VMs, flavors, ...
  86     It implements methods that does nothing and return ok"""
  87     def __init__(self, db, my_vims, db_vims, logger):
  88         self.db = db
  89         self.logger = logger
  90         self.my_vims = my_vims
  91         self.db_vims = db_vims
  92
  93     def new(self, ro_task, task_index, task_depends):
  94         return "BUILD", {}
  95
  96     def refresh(self, ro_task):
  97         """skip calling VIM to get image, flavor status. Assumes ok"""
  98         if ro_task["vim_info"]["vim_status"] == "VIM_ERROR":
  99             return "FAILED", {}
 100         return "DONE", {}
 101
 102     def delete(self, ro_task, task_index):
 103         """skip calling VIM to delete image. Assumes ok"""
 104         return "DONE", {}
 105
 106     def exec(self, ro_task, task_index, task_depends):
 107         return "DONE", None, None
 108
 109
 110 class VimInteractionNet(VimInteractionBase):
 111
 112     def new(self, ro_task, task_index, task_depends):
 113         vim_net_id = None
 114         task = ro_task["tasks"][task_index]
 115         task_id = task["task_id"]
 116         created = False
 117         created_items = {}
 118         target_vim = self.my_vims[ro_task["target_id"]]
 119         try:
 120             # FIND
 121             if task.get("find_params"):
 122                 # if management, get configuration of VIM
 123                 if task["find_params"].get("filter_dict"):
 124                     vim_filter = task["find_params"]["filter_dict"]
 125                 elif task["find_params"].get("mgmt"):   # mamagement network
 126                     if deep_get(self.db_vims[ro_task["target_id"]], "config", "management_network_id"):
 127                         vim_filter = {"id": self.db_vims[ro_task["target_id"]]["config"]["management_network_id"]}
 128                     elif deep_get(self.db_vims[ro_task["target_id"]], "config", "management_network_name"):
 129                         vim_filter = {"name": self.db_vims[ro_task["target_id"]]["config"]["management_network_name"]}
 130                     else:
 131                         vim_filter = {"name": task["find_params"]["name"]}
 132                 else:
 133                     raise NsWorkerExceptionNotFound("Invalid find_params for new_net {}".format(task["find_params"]))
 134
 135                 vim_nets = target_vim.get_network_list(vim_filter)
 136                 if not vim_nets and not task.get("params"):
 137                     raise NsWorkerExceptionNotFound("Network not found with this criteria: '{}'".format(
 138                         task.get("find_params")))
 139                 elif len(vim_nets) > 1:
 140                     raise NsWorkerException(
 141                         "More than one network found with this criteria: '{}'".format(task["find_params"]))
 142                 if vim_nets:
 143                     vim_net_id = vim_nets[0]["id"]
 144             else:
 145                 # CREATE
 146                 params = task["params"]
 147                 vim_net_id, created_items = target_vim.new_network(**params)
 148                 created = True
 149
 150             ro_vim_item_update = {"vim_id": vim_net_id,
 151                                   "vim_status": "BUILD",
 152                                   "created": created,
 153                                   "created_items": created_items,
 154                                   "vim_details": None}
 155             self.logger.debug(
 156                 "task={} {} new-net={} created={}".format(task_id, ro_task["target_id"], vim_net_id, created))
 157             return "BUILD", ro_vim_item_update
 158         except (vimconn.VimConnException, NsWorkerException) as e:
 159             self.logger.error("task={} vim={} new-net: {}".format(task_id, ro_task["target_id"], e))
 160             ro_vim_item_update = {"vim_status": "VIM_ERROR",
 161                                   "created": created,
 162                                   "vim_details": str(e)}
 163             return "FAILED", ro_vim_item_update
 164
 165     def refresh(self, ro_task):
 166         """Call VIM to get network status"""
 167         ro_task_id = ro_task["_id"]
 168         target_vim = self.my_vims[ro_task["target_id"]]
 169
 170         vim_id = ro_task["vim_info"]["vim_id"]
 171         net_to_refresh_list = [vim_id]
 172         try:
 173             vim_dict = target_vim.refresh_nets_status(net_to_refresh_list)
 174             vim_info = vim_dict[vim_id]
 175             if vim_info["status"] == "ACTIVE":
 176                 task_status = "DONE"
 177             elif vim_info["status"] == "BUILD":
 178                 task_status = "BUILD"
 179             else:
 180                 task_status = "FAILED"
 181         except vimconn.VimConnException as e:
 182             # Mark all tasks at VIM_ERROR status
 183             self.logger.error("ro_task={} vim={} get-net={}: {}".format(ro_task_id, ro_task["target_id"], vim_id, e))
 184             vim_info = {"status": "VIM_ERROR", "error_msg": str(e)}
 185             task_status = "FAILED"
 186
 187         ro_vim_item_update = {}
 188         if ro_task["vim_info"]["vim_status"] != vim_info["status"]:
 189             ro_vim_item_update["vim_status"] = vim_info["status"]
 190         if ro_task["vim_info"]["vim_name"] != vim_info.get("name"):
 191             ro_vim_item_update["vim_name"] = vim_info.get("name")
 192         if vim_info["status"] in ("ERROR", "VIM_ERROR"):
 193             if ro_task["vim_info"]["vim_details"] != vim_info.get("error_msg"):
 194                 ro_vim_item_update["vim_details"] = vim_info.get("error_msg")
 195         elif vim_info["status"] == "DELETED":
 196             ro_vim_item_update["vim_id"] = None
 197             ro_vim_item_update["vim_details"] = "Deleted externally"
 198         else:
 199             if ro_task["vim_info"]["vim_details"] != vim_info["vim_info"]:
 200                 ro_vim_item_update["vim_details"] = vim_info["vim_info"]
 201         if ro_vim_item_update:
 202             self.logger.debug("ro_task={} {} get-net={}: status={} {}".format(
 203                 ro_task_id, ro_task["target_id"], vim_id, ro_vim_item_update.get("vim_status"),
 204                 ro_vim_item_update.get("vim_details") if ro_vim_item_update.get("vim_status") != "ACTIVE" else ''))
 205         return task_status, ro_vim_item_update
 206
 207     def delete(self, ro_task, task_index):
 208         task = ro_task["tasks"][task_index]
 209         task_id = task["task_id"]
 210         net_vim_id = ro_task["vim_info"]["vim_id"]
 211         ro_vim_item_update_ok = {"vim_status": "DELETED",
 212                                  "created": False,
 213                                  "vim_details": "DELETED",
 214                                  "vim_id": None}
 215         try:
 216             if net_vim_id or ro_task["vim_info"]["created_items"]:
 217                 target_vim = self.my_vims[ro_task["target_id"]]
 218                 target_vim.delete_network(net_vim_id, ro_task["vim_info"]["created_items"])
 219
 220         except vimconn.VimConnNotFoundException:
 221             ro_vim_item_update_ok["vim_details"] = "already deleted"
 222
 223         except vimconn.VimConnException as e:
 224             self.logger.error("ro_task={} vim={} del-net={}: {}".format(ro_task["_id"], ro_task["target_id"],
 225                                                                         net_vim_id, e))
 226             ro_vim_item_update = {"vim_status": "VIM_ERROR",
 227                                   "vim_details": "Error while deleting: {}".format(e)}
 228             return "FAILED", ro_vim_item_update
 229
 230         self.logger.debug("task={} {} del-net={} {}".format(task_id, ro_task["target_id"], net_vim_id,
 231                                                             ro_vim_item_update_ok.get("vim_details", "")))
 232         return "DONE", ro_vim_item_update_ok
 233
 234
 235 class VimInteractionVdu(VimInteractionBase):
 236     max_retries_inject_ssh_key = 20    # 20 times
 237     time_retries_inject_ssh_key = 30   # wevery 30 seconds
 238
 239     def new(self, ro_task, task_index, task_depends):
 240         task = ro_task["tasks"][task_index]
 241         task_id = task["task_id"]
 242         created = False
 243         created_items = {}
 244         target_vim = self.my_vims[ro_task["target_id"]]
 245         try:
 246             created = True
 247             params = task["params"]
 248             params_copy = deepcopy(params)
 249             net_list = params_copy["net_list"]
 250             for net in net_list:
 251                 if "net_id" in net and net["net_id"].startswith("TASK-"):  # change task_id into network_id
 252                     network_id = task_depends[net["net_id"]]
 253                     if not network_id:
 254                         raise NsWorkerException("Cannot create VM because depends on a network not created or found "
 255                                                 "for {}".format(net["net_id"]))
 256                     net["net_id"] = network_id
 257             if params_copy["image_id"].startswith("TASK-"):
 258                 params_copy["image_id"] = task_depends[params_copy["image_id"]]
 259             if params_copy["flavor_id"].startswith("TASK-"):
 260                 params_copy["flavor_id"] = task_depends[params_copy["flavor_id"]]
 261
 262             vim_vm_id, created_items = target_vim.new_vminstance(**params_copy)
 263             interfaces = [iface["vim_id"] for iface in params_copy["net_list"]]
 264
 265             ro_vim_item_update = {"vim_id": vim_vm_id,
 266                                   "vim_status": "BUILD",
 267                                   "created": created,
 268                                   "created_items": created_items,
 269                                   "vim_details": None,
 270                                   "interfaces_vim_ids": interfaces,
 271                                   "interfaces": [],
 272                                   }
 273             self.logger.debug(
 274                 "task={} {} new-vm={} created={}".format(task_id, ro_task["target_id"], vim_vm_id, created))
 275             return "BUILD", ro_vim_item_update
 276         except (vimconn.VimConnException, NsWorkerException) as e:
 277             self.logger.error("task={} {} new-vm: {}".format(task_id, ro_task["target_id"], e))
 278             ro_vim_item_update = {"vim_status": "VIM_ERROR",
 279                                   "created": created,
 280                                   "vim_details": str(e)}
 281             return "FAILED", ro_vim_item_update
 282
 283     def delete(self, ro_task, task_index):
 284         task = ro_task["tasks"][task_index]
 285         task_id = task["task_id"]
 286         vm_vim_id = ro_task["vim_info"]["vim_id"]
 287         ro_vim_item_update_ok = {"vim_status": "DELETED",
 288                                  "created": False,
 289                                  "vim_details": "DELETED",
 290                                  "vim_id": None}
 291         try:
 292             if vm_vim_id or ro_task["vim_info"]["created_items"]:
 293                 target_vim = self.my_vims[ro_task["target_id"]]
 294                 target_vim.delete_vminstance(vm_vim_id, ro_task["vim_info"]["created_items"])
 295
 296         except vimconn.VimConnNotFoundException:
 297             ro_vim_item_update_ok["vim_details"] = "already deleted"
 298
 299         except vimconn.VimConnException as e:
 300             self.logger.error("ro_task={} vim={} del-vm={}: {}".format(ro_task["_id"], ro_task["target_id"],
 301                                                                        vm_vim_id, e))
 302             ro_vim_item_update = {"vim_status": "VIM_ERROR",
 303                                   "vim_details": "Error while deleting: {}".format(e)}
 304             return "FAILED", ro_vim_item_update
 305
 306         self.logger.debug("task={} {} del-vm={} {}".format(task_id, ro_task["target_id"], vm_vim_id,
 307                                                            ro_vim_item_update_ok.get("vim_details", "")))
 308         return "DONE", ro_vim_item_update_ok
 309
 310     def refresh(self, ro_task):
 311         """Call VIM to get vm status"""
 312         ro_task_id = ro_task["_id"]
 313         target_vim = self.my_vims[ro_task["target_id"]]
 314
 315         vim_id = ro_task["vim_info"]["vim_id"]
 316         if not vim_id:
 317             return None, None
 318         vm_to_refresh_list = [vim_id]
 319         try:
 320             vim_dict = target_vim.refresh_vms_status(vm_to_refresh_list)
 321             vim_info = vim_dict[vim_id]
 322             if vim_info["status"] == "ACTIVE":
 323                 task_status = "DONE"
 324             elif vim_info["status"] == "BUILD":
 325                 task_status = "BUILD"
 326             else:
 327                 task_status = "FAILED"
 328             # try to load and parse vim_information
 329             try:
 330                 vim_info_info = yaml.safe_load(vim_info["vim_info"])
 331                 if vim_info_info.get("name"):
 332                     vim_info["name"] = vim_info_info["name"]
 333             except Exception:
 334                 pass
 335         except vimconn.VimConnException as e:
 336             # Mark all tasks at VIM_ERROR status
 337             self.logger.error("ro_task={} vim={} get-vm={}: {}".format(ro_task_id, ro_task["target_id"], vim_id, e))
 338             vim_info = {"status": "VIM_ERROR", "error_msg": str(e)}
 339             task_status = "FAILED"
 340
 341         ro_vim_item_update = {}
 342         # Interfaces cannot be present if e.g. VM is not present, that is status=DELETED
 343         vim_interfaces = []
 344         if vim_info.get("interfaces"):
 345             for vim_iface_id in ro_task["vim_info"]["interfaces_vim_ids"]:
 346                 iface = next((iface for iface in vim_info["interfaces"] if vim_iface_id == iface["vim_interface_id"]),
 347                              None)
 348                 # if iface:
 349                 #     iface.pop("vim_info", None)
 350                 vim_interfaces.append(iface)
 351
 352         task_create = next(t for t in ro_task["tasks"] if t and t["action"] == "CREATE" and t["status"] != "FINISHED")
 353         if vim_interfaces and task_create.get("mgmt_vnf_interface") is not None:
 354             vim_interfaces[task_create["mgmt_vnf_interface"]]["mgmt_vnf_interface"] = True
 355         mgmt_vdu_iface = task_create.get("mgmt_vdu_interface", task_create.get("mgmt_vnf_interface", 0))
 356         if vim_interfaces:
 357             vim_interfaces[mgmt_vdu_iface]["mgmt_vdu_interface"] = True
 358
 359         if ro_task["vim_info"]["interfaces"] != vim_interfaces:
 360             ro_vim_item_update["interfaces"] = vim_interfaces
 361         if ro_task["vim_info"]["vim_status"] != vim_info["status"]:
 362             ro_vim_item_update["vim_status"] = vim_info["status"]
 363         if ro_task["vim_info"]["vim_name"] != vim_info.get("name"):
 364             ro_vim_item_update["vim_name"] = vim_info.get("name")
 365         if vim_info["status"] in ("ERROR", "VIM_ERROR"):
 366             if ro_task["vim_info"]["vim_details"] != vim_info.get("error_msg"):
 367                 ro_vim_item_update["vim_details"] = vim_info.get("error_msg")
 368         elif vim_info["status"] == "DELETED":
 369             ro_vim_item_update["vim_id"] = None
 370             ro_vim_item_update["vim_details"] = "Deleted externally"
 371         else:
 372             if ro_task["vim_info"]["vim_details"] != vim_info["vim_info"]:
 373                 ro_vim_item_update["vim_details"] = vim_info["vim_info"]
 374         if ro_vim_item_update:
 375             self.logger.debug("ro_task={} {} get-vm={}: status={} {}".format(
 376                 ro_task_id, ro_task["target_id"], vim_id, ro_vim_item_update.get("vim_status"),
 377                 ro_vim_item_update.get("vim_details") if ro_vim_item_update.get("vim_status") != "ACTIVE" else ''))
 378         return task_status, ro_vim_item_update
 379
 380     def exec(self, ro_task, task_index, task_depends):
 381         task = ro_task["tasks"][task_index]
 382         task_id = task["task_id"]
 383         target_vim = self.my_vims[ro_task["target_id"]]
 384         db_task_update = {"retries": 0}
 385         retries = task.get("retries", 0)
 386         try:
 387             params = task["params"]
 388             params_copy = deepcopy(params)
 389             params_copy["ro_key"] = self.db.decrypt(params_copy.pop("private_key"),
 390                                                     params_copy.pop("schema_version"), params_copy.pop("salt"))
 391             params_copy["ip_addr"] = params_copy.pop("ip_address")
 392             target_vim.inject_user_key(**params_copy)
 393             self.logger.debug(
 394                 "task={} {} action-vm=inject_key".format(task_id, ro_task["target_id"]))
 395             return "DONE", None, db_task_update,    # params_copy["key"]
 396         except (vimconn.VimConnException, NsWorkerException) as e:
 397             retries += 1
 398             if retries < self.max_retries_inject_ssh_key:
 399                 return "BUILD", None, {"retries": retries, "next_retry": self.time_retries_inject_ssh_key}
 400             self.logger.error("task={} {} inject-ssh-key: {}".format(task_id, ro_task["target_id"], e))
 401             ro_vim_item_update = {"vim_details": str(e)}
 402             return "FAILED", ro_vim_item_update, db_task_update
 403
 404
 405 class VimInteractionImage(VimInteractionBase):
 406
 407     def new(self, ro_task, task_index, task_depends):
 408         task = ro_task["tasks"][task_index]
 409         task_id = task["task_id"]
 410         created = False
 411         created_items = {}
 412         target_vim = self.my_vims[ro_task["target_id"]]
 413         try:
 414             # FIND
 415             if task.get("find_params"):
 416                 vim_images = target_vim.get_image_list(**task["find_params"])
 417                 if not vim_images:
 418                     raise NsWorkerExceptionNotFound("Image not found with this criteria: '{}'".format(
 419                         task["find_params"]))
 420                 elif len(vim_images) > 1:
 421                     raise NsWorkerException(
 422                         "More than one network found with this criteria: '{}'".format(task["find_params"]))
 423                 else:
 424                     vim_image_id = vim_images[0]["id"]
 425
 426             ro_vim_item_update = {"vim_id": vim_image_id,
 427                                   "vim_status": "DONE",
 428                                   "created": created,
 429                                   "created_items": created_items,
 430                                   "vim_details": None}
 431             self.logger.debug(
 432                 "task={} {} new-image={} created={}".format(task_id, ro_task["target_id"], vim_image_id, created))
 433             return "DONE", ro_vim_item_update
 434         except (NsWorkerException, vimconn.VimConnException) as e:
 435             self.logger.error("task={} {} new-image: {}".format(task_id, ro_task["target_id"], e))
 436             ro_vim_item_update = {"vim_status": "VIM_ERROR",
 437                                   "created": created,
 438                                   "vim_details": str(e)}
 439             return "FAILED", ro_vim_item_update
 440
 441
 442 class VimInteractionFlavor(VimInteractionBase):
 443
 444     def delete(self, ro_task, task_index):
 445         task = ro_task["tasks"][task_index]
 446         task_id = task["task_id"]
 447         flavor_vim_id = ro_task["vim_info"]["vim_id"]
 448         ro_vim_item_update_ok = {"vim_status": "DELETED",
 449                                  "created": False,
 450                                  "vim_details": "DELETED",
 451                                  "vim_id": None}
 452         try:
 453             if flavor_vim_id:
 454                 target_vim = self.my_vims[ro_task["target_id"]]
 455                 target_vim.delete_flavor(flavor_vim_id)
 456
 457         except vimconn.VimConnNotFoundException:
 458             ro_vim_item_update_ok["vim_details"] = "already deleted"
 459
 460         except vimconn.VimConnException as e:
 461             self.logger.error("ro_task={} vim={} del-flavor={}: {}".format(
 462                 ro_task["_id"], ro_task["target_id"], flavor_vim_id, e))
 463             ro_vim_item_update = {"vim_status": "VIM_ERROR",
 464                                   "vim_details": "Error while deleting: {}".format(e)}
 465             return "FAILED", ro_vim_item_update
 466
 467         self.logger.debug("task={} {} del-flavor={} {}".format(
 468             task_id, ro_task["target_id"], flavor_vim_id, ro_vim_item_update_ok.get("vim_details", "")))
 469         return "DONE", ro_vim_item_update_ok
 470
 471     def new(self, ro_task, task_index, task_depends):
 472         task = ro_task["tasks"][task_index]
 473         task_id = task["task_id"]
 474         created = False
 475         created_items = {}
 476         target_vim = self.my_vims[ro_task["target_id"]]
 477         try:
 478             # FIND
 479             vim_flavor_id = None
 480             if task.get("find_params"):
 481                 try:
 482                     flavor_data = task["find_params"]["flavor_data"]
 483                     vim_flavor_id = target_vim.get_flavor_id_from_data(flavor_data)
 484                 except vimconn.VimConnNotFoundException:
 485                     pass
 486
 487             if not vim_flavor_id and task.get("params"):
 488                 # CREATE
 489                 flavor_data = task["params"]["flavor_data"]
 490                 vim_flavor_id = target_vim.new_flavor(flavor_data)
 491                 created = True
 492
 493             ro_vim_item_update = {"vim_id": vim_flavor_id,
 494                                   "vim_status": "DONE",
 495                                   "created": created,
 496                                   "created_items": created_items,
 497                                   "vim_details": None}
 498             self.logger.debug(
 499                 "task={} {} new-flavor={} created={}".format(task_id, ro_task["target_id"], vim_flavor_id, created))
 500             return "DONE", ro_vim_item_update
 501         except (vimconn.VimConnException, NsWorkerException) as e:
 502             self.logger.error("task={} vim={} new-flavor: {}".format(task_id, ro_task["target_id"], e))
 503             ro_vim_item_update = {"vim_status": "VIM_ERROR",
 504                                   "created": created,
 505                                   "vim_details": str(e)}
 506             return "FAILED", ro_vim_item_update
 507
 508
 509 class VimInteractionSdnNet(VimInteractionBase):
 510
 511     @staticmethod
 512     def _match_pci(port_pci, mapping):
 513         """
 514         Check if port_pci matches with mapping
 515         mapping can have brackets to indicate that several chars are accepted. e.g
 516         pci '0000:af:10.1' matches with '0000:af:1[01].[1357]'
 517         :param port_pci: text
 518         :param mapping: text, can contain brackets to indicate several chars are available
 519         :return: True if matches, False otherwise
 520         """
 521         if not port_pci or not mapping:
 522             return False
 523         if port_pci == mapping:
 524             return True
 525
 526         mapping_index = 0
 527         pci_index = 0
 528         while True:
 529             bracket_start = mapping.find("[", mapping_index)
 530             if bracket_start == -1:
 531                 break
 532             bracket_end = mapping.find("]", bracket_start)
 533             if bracket_end == -1:
 534                 break
 535             length = bracket_start - mapping_index
 536             if length and port_pci[pci_index:pci_index + length] != mapping[mapping_index:bracket_start]:
 537                 return False
 538             if port_pci[pci_index + length] not in mapping[bracket_start+1:bracket_end]:
 539                 return False
 540             pci_index += length + 1
 541             mapping_index = bracket_end + 1
 542
 543         if port_pci[pci_index:] != mapping[mapping_index:]:
 544             return False
 545         return True
 546
 547     def _get_interfaces(self, vlds_to_connect, vim_account_id):
 548         """
 549         :param vlds_to_connect: list with format vnfrs:<id>:vld.<vld_id> or nsrs:<id>:vld.<vld_id>
 550         :param vim_account_id:
 551         :return:
 552         """
 553         interfaces = []
 554         for vld in vlds_to_connect:
 555             table, _, db_id = vld.partition(":")
 556             db_id, _, vld = db_id.partition(":")
 557             _, _, vld_id = vld.partition(".")
 558             if table == "vnfrs":
 559                 q_filter = {"vim-account-id": vim_account_id, "_id": db_id}
 560                 iface_key = "vnf-vld-id"
 561             else:  # table == "nsrs"
 562                 q_filter = {"vim-account-id": vim_account_id, "nsr-id-ref": db_id}
 563                 iface_key = "ns-vld-id"
 564             db_vnfrs = self.db.get_list("vnfrs", q_filter=q_filter)
 565             for db_vnfr in db_vnfrs:
 566                 for vdu_index, vdur in enumerate(db_vnfr.get("vdur", ())):
 567                     for iface_index, interface in enumerate(vdur["interfaces"]):
 568                         if interface.get(iface_key) == vld_id and \
 569                                 interface.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
 570                             # only SR-IOV o PT
 571                             interface_ = interface.copy()
 572                             interface_["id"] = "vnfrs:{}:vdu.{}.interfaces.{}".format(db_vnfr["_id"], vdu_index,
 573                                                                                       iface_index)
 574                             if vdur.get("status") == "ERROR":
 575                                 interface_["status"] = "ERROR"
 576                             interfaces.append(interface_)
 577         return interfaces
 578
 579     def refresh(self, ro_task):
 580         # look for task create
 581         task_create_index, _ = next(i_t for i_t in enumerate(ro_task["tasks"])
 582                                     if i_t[1] and i_t[1]["action"] == "CREATE" and i_t[1]["status"] != "FINISHED")
 583
 584         return self.new(ro_task, task_create_index, None)
 585
 586     def new(self, ro_task, task_index, task_depends):
 587
 588         task = ro_task["tasks"][task_index]
 589         task_id = task["task_id"]
 590         target_vim = self.my_vims[ro_task["target_id"]]
 591
 592         sdn_net_id = ro_task["vim_info"]["vim_id"]
 593
 594         created_items = ro_task["vim_info"].get("created_items")
 595         connected_ports = ro_task["vim_info"].get("connected_ports", [])
 596         new_connected_ports = []
 597         last_update = ro_task["vim_info"].get("last_update", 0)
 598         sdn_status = ro_task["vim_info"].get("vim_status", "BUILD") or "BUILD"
 599         error_list = []
 600         created = ro_task["vim_info"].get("created", False)
 601
 602         try:
 603
 604             # CREATE
 605             params = task["params"]
 606             vlds_to_connect = params["vlds"]
 607             associated_vim = params["target_vim"]
 608             additional_ports = params.get("sdn-ports") or ()  # external additional ports
 609             _, _, vim_account_id = associated_vim.partition(":")
 610             if associated_vim:
 611                 # get associated VIM
 612                 if associated_vim not in self.db_vims:
 613                     self.db_vims[associated_vim] = self.db.get_one("vim_accounts", {"_id": vim_account_id})
 614                 db_vim = self.db_vims[associated_vim]
 615
 616             # look for ports to connect
 617             ports = self._get_interfaces(vlds_to_connect, vim_account_id)
 618             # print(ports)
 619
 620             sdn_ports = []
 621             pending_ports = error_ports = 0
 622             vlan_used = None
 623             sdn_need_update = False
 624             for port in ports:
 625                 vlan_used = port.get("vlan") or vlan_used
 626                 # TODO. Do not connect if already done
 627                 if not port.get("compute_node") or not port.get("pci"):
 628                     if port.get("status") == "ERROR":
 629                         error_ports += 1
 630                     else:
 631                         pending_ports += 1
 632                     continue
 633                 pmap = None
 634                 compute_node_mappings = next((c for c in db_vim["config"].get("sdn-port-mapping", ())
 635                                              if c and c["compute_node"] == port["compute_node"]), None)
 636                 if compute_node_mappings:
 637                     # process port_mapping pci of type 0000:af:1[01].[1357]
 638                     pmap = next((p for p in compute_node_mappings["ports"]
 639                                  if self._match_pci(port["pci"], p.get("pci"))), None)
 640                 if not pmap:
 641                     if not db_vim["config"].get("mapping_not_needed"):
 642                         error_list.append("Port mapping not found for compute_node={} pci={}".format(
 643                             port["compute_node"], port["pci"]))
 644                         continue
 645                     pmap = {}
 646
 647                 service_endpoint_id = "{}:{}".format(port["compute_node"], port["pci"])
 648                 new_port = {
 649                     "service_endpoint_id": pmap.get("service_endpoint_id") or service_endpoint_id,
 650                     "service_endpoint_encapsulation_type": "dot1q" if port["type"] == "SR-IOV" else None,
 651                     "service_endpoint_encapsulation_info": {
 652                         "vlan": port.get("vlan"),
 653                         "mac": port.get("mac_address"),
 654                         "device_id": pmap.get("device_id") or port["compute_node"],  # device_id
 655                         "device_interface_id": pmap.get("device_interface_id") or port["pci"],
 656                         "switch_dpid": pmap.get("switch_id") or pmap.get("switch_dpid"),
 657                         "switch_port": pmap.get("switch_port"),
 658                         "service_mapping_info": pmap.get("service_mapping_info"),
 659                     }
 660                 }
 661
 662                 # TODO
 663                 # if port["modified_at"] > last_update:
 664                 #     sdn_need_update = True
 665                 new_connected_ports.append(port["id"])  # TODO
 666                 sdn_ports.append(new_port)
 667
 668             if error_ports:
 669                 error_list.append("{} interfaces have not been created as VDU is on ERROR status".format(error_ports))
 670
 671             # connect external ports
 672             for index, additional_port in enumerate(additional_ports):
 673                 additional_port_id = additional_port.get("service_endpoint_id") or "external-{}".format(index)
 674                 sdn_ports.append({
 675                     "service_endpoint_id": additional_port_id,
 676                     "service_endpoint_encapsulation_type": additional_port.get("service_endpoint_encapsulation_type",
 677                                                                                "dot1q"),
 678                     "service_endpoint_encapsulation_info": {
 679                         "vlan": additional_port.get("vlan") or vlan_used,
 680                         "mac": additional_port.get("mac_address"),
 681                         "device_id": additional_port.get("device_id"),
 682                         "device_interface_id": additional_port.get("device_interface_id"),
 683                         "switch_dpid": additional_port.get("switch_dpid") or additional_port.get("switch_id"),
 684                         "switch_port": additional_port.get("switch_port"),
 685                         "service_mapping_info": additional_port.get("service_mapping_info"),
 686                     }})
 687                 new_connected_ports.append(additional_port_id)
 688             sdn_info = ""
 689             # if there are more ports to connect or they have been modified, call create/update
 690             if error_list:
 691                 sdn_status = "ERROR"
 692                 sdn_info = "; ".join(error_list)
 693             elif set(connected_ports) != set(new_connected_ports) or sdn_need_update:
 694                 last_update = time.time()
 695                 if not sdn_net_id:
 696                     if len(sdn_ports) < 2:
 697                         sdn_status = "ACTIVE"
 698                         if not pending_ports:
 699                             self.logger.debug("task={} {} new-sdn-net done, less than 2 ports".
 700                                               format(task_id, ro_task["target_id"]))
 701                     else:
 702                         net_type = params.get("type") or "ELAN"
 703                         sdn_net_id, created_items = target_vim.create_connectivity_service(
 704                             net_type, sdn_ports)
 705                         created = True
 706                         self.logger.debug("task={} {} new-sdn-net={} created={}".
 707                                           format(task_id, ro_task["target_id"], sdn_net_id, created))
 708                 else:
 709                     created_items = target_vim.edit_connectivity_service(
 710                         sdn_net_id, conn_info=created_items, connection_points=sdn_ports)
 711                     created = True
 712                     self.logger.debug("task={} {} update-sdn-net={} created={}".
 713                                       format(task_id, ro_task["target_id"], sdn_net_id, created))
 714                 connected_ports = new_connected_ports
 715             elif sdn_net_id:
 716                 wim_status_dict = target_vim.get_connectivity_service_status(sdn_net_id, conn_info=created_items)
 717                 sdn_status = wim_status_dict["sdn_status"]
 718                 if wim_status_dict.get("sdn_info"):
 719                     sdn_info = str(wim_status_dict.get("sdn_info")) or ""
 720                 if wim_status_dict.get("error_msg"):
 721                     sdn_info = wim_status_dict.get("error_msg") or ""
 722
 723             if pending_ports:
 724                 if sdn_status != "ERROR":
 725                     sdn_info = "Waiting for getting interfaces location from VIM. Obtained '{}' of {}".format(
 726                         len(ports)-pending_ports, len(ports))
 727                 if sdn_status == "ACTIVE":
 728                     sdn_status = "BUILD"
 729
 730             ro_vim_item_update = {"vim_id": sdn_net_id,
 731                                   "vim_status": sdn_status,
 732                                   "created": created,
 733                                   "created_items": created_items,
 734                                   "connected_ports": connected_ports,
 735                                   "vim_details": sdn_info,
 736                                   "last_update": last_update}
 737             return sdn_status, ro_vim_item_update
 738         except Exception as e:
 739             self.logger.error("task={} vim={} new-net: {}".format(task_id, ro_task["target_id"], e),
 740                               exc_info=not isinstance(e, (sdnconn.SdnConnectorError, vimconn.VimConnException)))
 741             ro_vim_item_update = {"vim_status": "VIM_ERROR",
 742                                   "created": created,
 743                                   "vim_details": str(e)}
 744             return "FAILED", ro_vim_item_update
 745
 746     def delete(self, ro_task, task_index):
 747         task = ro_task["tasks"][task_index]
 748         task_id = task["task_id"]
 749         sdn_vim_id = ro_task["vim_info"].get("vim_id")
 750         ro_vim_item_update_ok = {"vim_status": "DELETED",
 751                                  "created": False,
 752                                  "vim_details": "DELETED",
 753                                  "vim_id": None}
 754         try:
 755             if sdn_vim_id:
 756                 target_vim = self.my_vims[ro_task["target_id"]]
 757                 target_vim.delete_connectivity_service(sdn_vim_id, ro_task["vim_info"].get("created_items"))
 758
 759         except Exception as e:
 760             if isinstance(e, sdnconn.SdnConnectorError) and e.http_code == HTTPStatus.NOT_FOUND.value:
 761                 ro_vim_item_update_ok["vim_details"] = "already deleted"
 762             else:
 763                 self.logger.error("ro_task={} vim={} del-sdn-net={}: {}".format(ro_task["_id"], ro_task["target_id"],
 764                                                                                 sdn_vim_id, e),
 765                                   exc_info=not isinstance(e, (sdnconn.SdnConnectorError, vimconn.VimConnException)))
 766                 ro_vim_item_update = {"vim_status": "VIM_ERROR",
 767                                       "vim_details": "Error while deleting: {}".format(e)}
 768                 return "FAILED", ro_vim_item_update
 769
 770         self.logger.debug("task={} {} del-sdn-net={} {}".format(task_id, ro_task["target_id"], sdn_vim_id,
 771                                                                 ro_vim_item_update_ok.get("vim_details", "")))
 772         return "DONE", ro_vim_item_update_ok
 773
 774
 775 class NsWorker(threading.Thread):
 776     REFRESH_BUILD = 5  # 5 seconds
 777     REFRESH_ACTIVE = 60  # 1 minute
 778     REFRESH_ERROR = 600
 779     REFRESH_IMAGE = 3600 * 10
 780     REFRESH_DELETE = 3600 * 10
 781     QUEUE_SIZE = 100
 782     terminate = False
 783
 784     def __init__(self, worker_index, config, plugins, db):
 785         """
 786
 787         :param worker_index: thread index
 788         :param config: general configuration of RO, among others the process_id with the docker id where it runs
 789         :param plugins: global shared dict with the loaded plugins
 790         :param db: database class instance to use
 791         """
 792         threading.Thread.__init__(self)
 793         self.config = config
 794         self.plugins = plugins
 795         self.plugin_name = "unknown"
 796         self.logger = logging.getLogger('ro.worker{}'.format(worker_index))
 797         self.worker_index = worker_index
 798         self.task_queue = queue.Queue(self.QUEUE_SIZE)
 799         self.my_vims = {}   # targetvim: vimplugin class
 800         self.db_vims = {}   # targetvim: vim information from database
 801         self.vim_targets = []   # targetvim list
 802         self.my_id = config["process_id"] + ":" + str(worker_index)
 803         self.db = db
 804         self.item2class = {
 805             "net": VimInteractionNet(self.db, self.my_vims, self.db_vims, self.logger),
 806             "vdu": VimInteractionVdu(self.db, self.my_vims, self.db_vims, self.logger),
 807             "image": VimInteractionImage(self.db, self.my_vims, self.db_vims, self.logger),
 808             "flavor": VimInteractionFlavor(self.db, self.my_vims, self.db_vims, self.logger),
 809             "sdn_net": VimInteractionSdnNet(self.db, self.my_vims, self.db_vims, self.logger),
 810         }
 811         self.time_last_task_processed = None
 812         self.tasks_to_delete = []  # lists of tasks to delete because nsrs or vnfrs has been deleted from db
 813         self.idle = True  # it is idle when there are not vim_targets associated
 814         self.task_locked_time = config["global"]["task_locked_time"]
 815
 816     def insert_task(self, task):
 817         try:
 818             self.task_queue.put(task, False)
 819             return None
 820         except queue.Full:
 821             raise NsWorkerException("timeout inserting a task")
 822
 823     def terminate(self):
 824         self.insert_task("exit")
 825
 826     def del_task(self, task):
 827         with self.task_lock:
 828             if task["status"] == "SCHEDULED":
 829                 task["status"] = "SUPERSEDED"
 830                 return True
 831             else:  # task["status"] == "processing"
 832                 self.task_lock.release()
 833                 return False
 834
 835     def _process_vim_config(self, target_id, db_vim):
 836         """
 837         Process vim config, creating vim configuration files as ca_cert
 838         :param target_id: vim/sdn/wim + id
 839         :param db_vim: Vim dictionary obtained from database
 840         :return: None. Modifies vim. Creates a folder target_id:worker_index and several files
 841         """
 842         if not db_vim.get("config"):
 843             return
 844         file_name = ""
 845         try:
 846             if db_vim["config"].get("ca_cert_content"):
 847                 file_name = "{}:{}".format(target_id, self.worker_index)
 848                 try:
 849                     mkdir(file_name)
 850                 except FileExistsError:
 851                     pass
 852                 file_name = file_name + "/ca_cert"
 853                 with open(file_name, "w") as f:
 854                     f.write(db_vim["config"]["ca_cert_content"])
 855                     del db_vim["config"]["ca_cert_content"]
 856                     db_vim["config"]["ca_cert"] = file_name
 857         except Exception as e:
 858             raise NsWorkerException("Error writing to file '{}': {}".format(file_name, e))
 859
 860     def _load_plugin(self, name, type="vim"):
 861         # type can be vim or sdn
 862         if "rovim_dummy" not in self.plugins:
 863             self.plugins["rovim_dummy"] = VimDummyConnector
 864         if "rosdn_dummy" not in self.plugins:
 865             self.plugins["rosdn_dummy"] = SdnDummyConnector
 866         if name in self.plugins:
 867             return self.plugins[name]
 868         try:
 869             for v in iter_entry_points('osm_ro{}.plugins'.format(type), name):
 870                 self.plugins[name] = v.load()
 871         except Exception as e:
 872             raise NsWorkerException("Cannot load plugin osm_{}: {}".format(name, e))
 873         if name and name not in self.plugins:
 874             raise NsWorkerException("Plugin 'osm_{n}' has not been installed".format(n=name))
 875         return self.plugins[name]
 876
 877     def _unload_vim(self, target_id):
 878         """
 879         Unload a vim_account. Removes it from self db_vims dictionary, my_vims dictionary and vim_targets list
 880         :param target_id: Contains type:_id; where type can be 'vim', ...
 881         :return: None.
 882         """
 883         try:
 884             self.db_vims.pop(target_id, None)
 885             self.my_vims.pop(target_id, None)
 886             if target_id in self.vim_targets:
 887                 self.vim_targets.remove(target_id)
 888             self.logger.info("Unloaded {}".format(target_id))
 889             rmtree("{}:{}".format(target_id, self.worker_index))
 890         except FileNotFoundError:
 891             pass  # this is raised by rmtree if folder does not exist
 892         except Exception as e:
 893             self.logger.error("Cannot unload {}: {}".format(target_id, e))
 894
 895     def _check_vim(self, target_id):
 896         """
 897         Load a VIM/SDN/WIM (if not loaded) and check connectivity, updating database with ENABLE or ERROR
 898         :param target_id: Contains type:_id; type can be 'vim', 'sdn' or 'wim'
 899         :return: None.
 900         """
 901         target, _, _id = target_id.partition(":")
 902         now = time.time()
 903         update_dict = {}
 904         unset_dict = {}
 905         op_text = ""
 906         step = ""
 907         loaded = target_id in self.vim_targets
 908         target_database = "vim_accounts" if target == "vim" else "wim_accounts" if target == "wim" else "sdns"
 909         try:
 910             step = "Getting {} from db".format(target_id)
 911             db_vim = self.db.get_one(target_database, {"_id": _id})
 912             for op_index, operation in enumerate(db_vim["_admin"].get("operations", ())):
 913                 if operation["operationState"] != "PROCESSING":
 914                     continue
 915                 locked_at = operation.get("locked_at")
 916                 if locked_at is not None and locked_at >= now - self.task_locked_time:
 917                     # some other thread is doing this operation
 918                     return
 919                 # lock
 920                 op_text = "_admin.operations.{}.".format(op_index)
 921                 if not self.db.set_one(target_database,
 922                                        q_filter={"_id": _id,
 923                                                  op_text + "operationState": "PROCESSING",
 924                                                  op_text + "locked_at": locked_at
 925                                                  },
 926                                        update_dict={op_text + "locked_at": now,
 927                                                     "admin.current_operation": op_index},
 928                                        fail_on_empty=False):
 929                     return
 930                 unset_dict[op_text + "locked_at"] = None
 931                 unset_dict["current_operation"] = None
 932                 step = "Loading " + target_id
 933                 error_text = self._load_vim(target_id)
 934                 if not error_text:
 935                     step = "Checking connectivity"
 936                     if target == 'vim':
 937                         self.my_vims[target_id].check_vim_connectivity()
 938                     else:
 939                         self.my_vims[target_id].check_credentials()
 940                 update_dict["_admin.operationalState"] = "ENABLED"
 941                 update_dict["_admin.detailed-status"] = ""
 942                 unset_dict[op_text + "detailed-status"] = None
 943                 update_dict[op_text + "operationState"] = "COMPLETED"
 944                 return
 945
 946         except Exception as e:
 947             error_text = "{}: {}".format(step, e)
 948             self.logger.error("{} for {}: {}".format(step, target_id, e))
 949
 950         finally:
 951             if update_dict or unset_dict:
 952                 if error_text:
 953                     update_dict[op_text + "operationState"] = "FAILED"
 954                     update_dict[op_text + "detailed-status"] = error_text
 955                     unset_dict.pop(op_text + "detailed-status", None)
 956                     update_dict["_admin.operationalState"] = "ERROR"
 957                     update_dict["_admin.detailed-status"] = error_text
 958                 if op_text:
 959                     update_dict[op_text + "statusEnteredTime"] = now
 960                 self.db.set_one(target_database, q_filter={"_id": _id}, update_dict=update_dict, unset=unset_dict,
 961                                 fail_on_empty=False)
 962             if not loaded:
 963                 self._unload_vim(target_id)
 964
 965     def _reload_vim(self, target_id):
 966         if target_id in self.vim_targets:
 967             self._load_vim(target_id)
 968         else:
 969             # if the vim is not loaded, but database information of VIM is cached at self.db_vims,
 970             # just remove it to force load again next time it is needed
 971             self.db_vims.pop(target_id, None)
 972
 973     def _load_vim(self, target_id):
 974         """
 975         Load or reload a vim_account, sdn_controller or wim_account.
 976         Read content from database, load the plugin if not loaded.
 977         In case of error loading the plugin, it load a failing VIM_connector
 978         It fills self db_vims dictionary, my_vims dictionary and vim_targets list
 979         :param target_id: Contains type:_id; where type can be 'vim', ...
 980         :return: None if ok, descriptive text if error
 981         """
 982         target, _, _id = target_id.partition(":")
 983         target_database = "vim_accounts" if target == "vim" else "wim_accounts" if target == "wim" else "sdns"
 984         plugin_name = ""
 985         vim = None
 986         try:
 987             step = "Getting {}={} from db".format(target, _id)
 988             # TODO process for wim, sdnc, ...
 989             vim = self.db.get_one(target_database, {"_id": _id})
 990
 991             # if deep_get(vim, "config", "sdn-controller"):
 992             #     step = "Getting sdn-controller-id='{}' from db".format(vim["config"]["sdn-controller"])
 993             #     db_sdn = self.db.get_one("sdns", {"_id": vim["config"]["sdn-controller"]})
 994
 995             step = "Decrypting password"
 996             schema_version = vim.get("schema_version")
 997             self.db.encrypt_decrypt_fields(vim, "decrypt", fields=('password', 'secret'),
 998                                            schema_version=schema_version, salt=_id)
 999             self._process_vim_config(target_id, vim)
1000             if target == "vim":
1001                 plugin_name = "rovim_" + vim["vim_type"]
1002                 step = "Loading plugin '{}'".format(plugin_name)
1003                 vim_module_conn = self._load_plugin(plugin_name)
1004                 step = "Loading {}'".format(target_id)
1005                 self.my_vims[target_id] = vim_module_conn(
1006                     uuid=vim['_id'], name=vim['name'],
1007                     tenant_id=vim.get('vim_tenant_id'), tenant_name=vim.get('vim_tenant_name'),
1008                     url=vim['vim_url'], url_admin=None,
1009                     user=vim['vim_user'], passwd=vim['vim_password'],
1010                     config=vim.get('config') or {}, persistent_info={}
1011                 )
1012             else:  # sdn
1013                 plugin_name = "rosdn_" + vim["type"]
1014                 step = "Loading plugin '{}'".format(plugin_name)
1015                 vim_module_conn = self._load_plugin(plugin_name, "sdn")
1016                 step = "Loading {}'".format(target_id)
1017                 wim = deepcopy(vim)
1018                 wim_config = wim.pop("config", {}) or {}
1019                 wim["uuid"] = wim["_id"]
1020                 wim["wim_url"] = wim["url"]
1021                 if wim.get("dpid"):
1022                     wim_config["dpid"] = wim.pop("dpid")
1023                 if wim.get("switch_id"):
1024                     wim_config["switch_id"] = wim.pop("switch_id")
1025                 self.my_vims[target_id] = vim_module_conn(wim, wim, wim_config)  # wim, wim_account, config
1026             self.db_vims[target_id] = vim
1027             self.error_status = None
1028             self.logger.info("Connector loaded for {}, plugin={}".format(target_id, plugin_name))
1029         except Exception as e:
1030             self.logger.error("Cannot load {} plugin={}: {} {}".format(
1031                 target_id, plugin_name, step, e))
1032             self.db_vims[target_id] = vim or {}
1033             self.db_vims[target_id] = FailingConnector(str(e))
1034             error_status = "{} Error: {}".format(step, e)
1035             return error_status
1036         finally:
1037             if target_id not in self.vim_targets:
1038                 self.vim_targets.append(target_id)
1039
1040     def _get_db_task(self):
1041         """
1042         Read actions from database and reload them at memory. Fill self.refresh_list, pending_list, vim_actions
1043         :return: None
1044         """
1045         now = time.time()
1046         if not self.time_last_task_processed:
1047             self.time_last_task_processed = now
1048         try:
1049             while True:
1050                 locked = self.db.set_one(
1051                     "ro_tasks",
1052                     q_filter={"target_id": self.vim_targets,
1053                               "tasks.status": ['SCHEDULED', 'BUILD', 'DONE', 'FAILED'],
1054                               "locked_at.lt": now - self.task_locked_time,
1055                               "to_check_at.lt": self.time_last_task_processed},
1056                     update_dict={"locked_by": self.my_id, "locked_at": now},
1057                     fail_on_empty=False)
1058                 if locked:
1059                     # read and return
1060                     ro_task = self.db.get_one(
1061                         "ro_tasks",
1062                         q_filter={"target_id": self.vim_targets,
1063                                   "tasks.status": ['SCHEDULED', 'BUILD', 'DONE', 'FAILED'],
1064                                   "locked_at": now})
1065                     return ro_task
1066                 if self.time_last_task_processed == now:
1067                     self.time_last_task_processed = None
1068                     return None
1069                 else:
1070                     self.time_last_task_processed = now
1071                     # self.time_last_task_processed = min(self.time_last_task_processed + 1000, now)
1072
1073         except DbException as e:
1074             self.logger.error("Database exception at _get_db_task: {}".format(e))
1075         except Exception as e:
1076             self.logger.critical("Unexpected exception at _get_db_task: {}".format(e), exc_info=True)
1077         return None
1078
1079     def _delete_task(self, ro_task, task_index, task_depends, db_update):
1080         """
1081         Determine if this task need to be done or superseded
1082         :return: None
1083         """
1084         my_task = ro_task["tasks"][task_index]
1085         task_id = my_task["task_id"]
1086         needed_delete = ro_task["vim_info"]["created"] or ro_task["vim_info"].get("created_items", False)
1087         if my_task["status"] == "FAILED":
1088             return None, None  # TODO need to be retry??
1089         try:
1090             for index, task in enumerate(ro_task["tasks"]):
1091                 if index == task_index or not task:
1092                     continue  # own task
1093                 if my_task["target_record"] == task["target_record"] and task["action"] == "CREATE":
1094                     # set to finished
1095                     db_update["tasks.{}.status".format(index)] = task["status"] = "FINISHED"
1096                 elif task["action"] == "CREATE" and task["status"] not in ("FINISHED", "SUPERSEDED"):
1097                     needed_delete = False
1098             if needed_delete:
1099                 return self.item2class[my_task["item"]].delete(ro_task, task_index)
1100             else:
1101                 return "SUPERSEDED", None
1102         except Exception as e:
1103             if not isinstance(e, NsWorkerException):
1104                 self.logger.critical("Unexpected exception at _delete_task task={}: {}".format(task_id, e),
1105                                      exc_info=True)
1106             return "FAILED", {"vim_status": "VIM_ERROR", "vim_details": str(e)}
1107
1108     def _create_task(self, ro_task, task_index, task_depends, db_update):
1109         """
1110         Determine if this task need to create something at VIM
1111         :return: None
1112         """
1113         my_task = ro_task["tasks"][task_index]
1114         task_id = my_task["task_id"]
1115         task_status = None
1116         if my_task["status"] == "FAILED":
1117             return None, None  # TODO need to be retry??
1118         elif my_task["status"] == "SCHEDULED":
1119             # check if already created by another task
1120             for index, task in enumerate(ro_task["tasks"]):
1121                 if index == task_index or not task:
1122                     continue  # own task
1123                 if task["action"] == "CREATE" and task["status"] not in ("SCHEDULED", "FINISHED", "SUPERSEDED"):
1124                     return task["status"], "COPY_VIM_INFO"
1125
1126             try:
1127                 task_status, ro_vim_item_update = self.item2class[my_task["item"]].new(
1128                     ro_task, task_index, task_depends)
1129                 # TODO update other CREATE tasks
1130             except Exception as e:
1131                 if not isinstance(e, NsWorkerException):
1132                     self.logger.error("Error executing task={}: {}".format(task_id, e), exc_info=True)
1133                 task_status = "FAILED"
1134                 ro_vim_item_update = {"vim_status": "VIM_ERROR", "vim_details": str(e)}
1135                 # TODO update    ro_vim_item_update
1136             return task_status, ro_vim_item_update
1137         else:
1138             return None, None
1139
1140     def _get_dependency(self, task_id, ro_task=None, target_id=None):
1141         """
1142         Look for dependency task
1143         :param task_id: Can be one of
1144             1. target_vim+blank+task.target_record_id: "(vim|sdn|wim):<id> (vnfrs|nsrs):(vld|vdu|flavor|image).<id>"
1145             2. task.target_record_id: "(vnfrs|nsrs):(vld|vdu|flavor|image).<id>"
1146             3. task.task_id: "<action_id>:number"
1147         :param ro_task:
1148         :param target_id:
1149         :return: database ro_task plus index of task
1150         """
1151         if task_id.startswith("vim:") or task_id.startswith("sdn:") or task_id.startswith("wim:"):
1152             target_id, _, task_id = task_id.partition(" ")
1153
1154         if task_id.startswith("nsrs:") or task_id.startswith("vnfrs:"):
1155             ro_task_dependency = self.db.get_one(
1156                 "ro_tasks",
1157                 q_filter={"target_id": target_id,
1158                           "tasks.target_record_id": task_id
1159                           },
1160                 fail_on_empty=False)
1161             if ro_task_dependency:
1162                 for task_index, task in enumerate(ro_task_dependency["tasks"]):
1163                     if task["target_record_id"] == task_id:
1164                         return ro_task_dependency, task_index
1165
1166         else:
1167             if ro_task:
1168                 for task_index, task in enumerate(ro_task["tasks"]):
1169                     if task and task["task_id"] == task_id:
1170                         return ro_task, task_index
1171             ro_task_dependency = self.db.get_one(
1172                 "ro_tasks",
1173                 q_filter={"tasks.ANYINDEX.task_id": task_id,
1174                           "tasks.ANYINDEX.target_record.ne": None
1175                           },
1176                 fail_on_empty=False)
1177             if ro_task_dependency:
1178                 for task_index, task in ro_task_dependency["tasks"]:
1179                     if task["task_id"] == task_id:
1180                         return ro_task_dependency, task_index
1181         raise NsWorkerException("Cannot get depending task {}".format(task_id))
1182
1183     def _process_pending_tasks(self, ro_task):
1184         ro_task_id = ro_task["_id"]
1185         now = time.time()
1186         next_check_at = now + (24*60*60)   # one day
1187         db_ro_task_update = {}
1188
1189         def _update_refresh(new_status):
1190             # compute next_refresh
1191             nonlocal task
1192             nonlocal next_check_at
1193             nonlocal db_ro_task_update
1194             nonlocal ro_task
1195
1196             next_refresh = time.time()
1197             if task["item"] in ("image", "flavor"):
1198                 next_refresh += self.REFRESH_IMAGE
1199             elif new_status == "BUILD":
1200                 next_refresh += self.REFRESH_BUILD
1201             elif new_status == "DONE":
1202                 next_refresh += self.REFRESH_ACTIVE
1203             else:
1204                 next_refresh += self.REFRESH_ERROR
1205             next_check_at = min(next_check_at, next_refresh)
1206             db_ro_task_update["vim_info.refresh_at"] = next_refresh
1207             ro_task["vim_info"]["refresh_at"] = next_refresh
1208
1209         try:
1210             # 0: get task_status_create
1211             lock_object = None
1212             task_status_create = None
1213             task_create = next((t for t in ro_task["tasks"] if t and t["action"] == "CREATE" and
1214                                 t["status"] in ("BUILD", "DONE")), None)
1215             if task_create:
1216                 task_status_create = task_create["status"]
1217             # 1: look for tasks in status SCHEDULED, or in status CREATE if action is  DONE or BUILD
1218             for task_action in ("DELETE", "CREATE", "EXEC"):
1219                 db_vim_update = None
1220                 new_status = None
1221                 for task_index, task in enumerate(ro_task["tasks"]):
1222                     if not task:
1223                         continue  # task deleted
1224                     task_depends = {}
1225                     target_update = None
1226                     if (task_action in ("DELETE", "EXEC") and task["status"] not in ("SCHEDULED", "BUILD")) or \
1227                             task["action"] != task_action or \
1228                             (task_action == "CREATE" and task["status"] in ("FINISHED", "SUPERSEDED")):
1229                         continue
1230                     task_path = "tasks.{}.status".format(task_index)
1231                     try:
1232                         db_vim_info_update = None
1233                         if task["status"] == "SCHEDULED":
1234                             # check if tasks that this depends on have been completed
1235                             dependency_not_completed = False
1236                             for dependency_task_id in (task.get("depends_on") or ()):
1237                                 dependency_ro_task, dependency_task_index = \
1238                                     self._get_dependency(dependency_task_id, target_id=ro_task["target_id"])
1239                                 dependency_task = dependency_ro_task["tasks"][dependency_task_index]
1240                                 if dependency_task["status"] == "SCHEDULED":
1241                                     dependency_not_completed = True
1242                                     next_check_at = min(next_check_at, dependency_ro_task["to_check_at"])
1243                                     break
1244                                 elif dependency_task["status"] == "FAILED":
1245                                     error_text = "Cannot {} {} because depends on failed {} {} id={}): {}".format(
1246                                         task["action"], task["item"], dependency_task["action"],
1247                                         dependency_task["item"], dependency_task_id,
1248                                         dependency_ro_task["vim_info"].get("vim_details"))
1249                                     self.logger.error("task={} {}".format(task["task_id"], error_text))
1250                                     raise NsWorkerException(error_text)
1251
1252                                 task_depends[dependency_task_id] = dependency_ro_task["vim_info"]["vim_id"]
1253                                 task_depends["TASK-{}".format(dependency_task_id)] = \
1254                                     dependency_ro_task["vim_info"]["vim_id"]
1255                             if dependency_not_completed:
1256                                 # TODO set at vim_info.vim_details that it is waiting
1257                                 continue
1258                         # before calling VIM-plugin as it can take more than task_locked_time, insert to LockRenew
1259                         # the task of renew this locking. It will update database locket_at periodically
1260                         if not lock_object:
1261                             lock_object = LockRenew.add_lock_object("ro_tasks", ro_task, self)
1262                         if task["action"] == "DELETE":
1263                             new_status, db_vim_info_update = self._delete_task(ro_task, task_index,
1264                                                                                task_depends, db_ro_task_update)
1265                             new_status = "FINISHED" if new_status == "DONE" else new_status
1266                             # ^with FINISHED instead of DONE it will not be refreshing
1267                             if new_status in ("FINISHED", "SUPERSEDED"):
1268                                 target_update = "DELETE"
1269                         elif task["action"] == "EXEC":
1270                             new_status, db_vim_info_update, db_task_update = self.item2class[task["item"]].exec(
1271                                 ro_task, task_index, task_depends)
1272                             new_status = "FINISHED" if new_status == "DONE" else new_status
1273                             # ^with FINISHED instead of DONE it will not be refreshing
1274                             if db_task_update:
1275                                 # load into database the modified db_task_update "retries" and "next_retry"
1276                                 if db_task_update.get("retries"):
1277                                     db_ro_task_update["tasks.{}.retries".format(task_index)] = db_task_update["retries"]
1278                                 next_check_at = time.time() + db_task_update.get("next_retry", 60)
1279                             target_update = None
1280                         elif task["action"] == "CREATE":
1281                             if task["status"] == "SCHEDULED":
1282                                 if task_status_create:
1283                                     new_status = task_status_create
1284                                     target_update = "COPY_VIM_INFO"
1285                                 else:
1286                                     new_status, db_vim_info_update = \
1287                                         self.item2class[task["item"]].new(ro_task, task_index, task_depends)
1288                                     # self._create_task(ro_task, task_index, task_depends, db_ro_task_update)
1289                                     _update_refresh(new_status)
1290                             else:
1291                                 if ro_task["vim_info"]["refresh_at"] and now > ro_task["vim_info"]["refresh_at"]:
1292                                     new_status, db_vim_info_update = self.item2class[task["item"]].refresh(ro_task)
1293                                     _update_refresh(new_status)
1294                     except Exception as e:
1295                         new_status = "FAILED"
1296                         db_vim_info_update = {"vim_status": "VIM_ERROR", "vim_details": str(e)}
1297                         if not isinstance(e, (NsWorkerException, vimconn.VimConnException)):
1298                             self.logger.error("Unexpected exception at _delete_task task={}: {}".
1299                                               format(task["task_id"], e), exc_info=True)
1300
1301                     try:
1302                         if db_vim_info_update:
1303                             db_vim_update = db_vim_info_update.copy()
1304                             db_ro_task_update.update({"vim_info." + k: v for k, v in db_vim_info_update.items()})
1305                             ro_task["vim_info"].update(db_vim_info_update)
1306
1307                         if new_status:
1308                             if task_action == "CREATE":
1309                                 task_status_create = new_status
1310                             db_ro_task_update[task_path] = new_status
1311                         if target_update or db_vim_update:
1312
1313                             if target_update == "DELETE":
1314                                 self._update_target(task, None)
1315                             elif target_update == "COPY_VIM_INFO":
1316                                 self._update_target(task, ro_task["vim_info"])
1317                             else:
1318                                 self._update_target(task, db_vim_update)
1319
1320                     except Exception as e:
1321                         if isinstance(e, DbException) and e.http_code == HTTPStatus.NOT_FOUND:
1322                             # if the vnfrs or nsrs has been removed from database, this task must be removed
1323                             self.logger.debug("marking to delete task={}".format(task["task_id"]))
1324                             self.tasks_to_delete.append(task)
1325                         else:
1326                             self.logger.error("Unexpected exception at _update_target task={}: {}".
1327                                               format(task["task_id"], e), exc_info=True)
1328
1329             locked_at = ro_task["locked_at"]
1330             if lock_object:
1331                 locked_at = [lock_object["locked_at"], lock_object["locked_at"] + self.task_locked_time]
1332                 # locked_at contains two times to avoid race condition. In case the lock has been renew, it will
1333                 # contain exactly locked_at + self.task_locked_time
1334                 LockRenew.remove_lock_object(lock_object)
1335             q_filter = {"_id": ro_task["_id"], "to_check_at": ro_task["to_check_at"], "locked_at": locked_at}
1336             # modify own task. Try filtering by to_next_check. For race condition if to_check_at has been modified,
1337             # outside this task (by ro_nbi) do not update it
1338             db_ro_task_update["locked_by"] = None
1339             # locked_at converted to int only for debugging. When has not decimals it means it has been unlocked
1340             db_ro_task_update["locked_at"] = int(now - self.task_locked_time)
1341             db_ro_task_update["modified_at"] = now
1342             db_ro_task_update["to_check_at"] = next_check_at
1343             if not self.db.set_one("ro_tasks",
1344                                    update_dict=db_ro_task_update,
1345                                    q_filter=q_filter,
1346                                    fail_on_empty=False):
1347                 del db_ro_task_update["to_check_at"]
1348                 del q_filter["to_check_at"]
1349                 self.db.set_one("ro_tasks",
1350                                 q_filter=q_filter,
1351                                 update_dict=db_ro_task_update,
1352                                 fail_on_empty=True)
1353         except DbException as e:
1354             self.logger.error("ro_task={} Error updating database {}".format(ro_task_id, e))
1355         except Exception as e:
1356             self.logger.error("Error executing ro_task={}: {}".format(ro_task_id, e), exc_info=True)
1357
1358     def _update_target(self, task, ro_vim_item_update):
1359         table, _, temp = task["target_record"].partition(":")
1360         _id, _, path_vim_status = temp.partition(":")
1361         path_item = path_vim_status[:path_vim_status.rfind(".")]
1362         path_item = path_item[:path_item.rfind(".")]
1363         # path_vim_status: dot separated list targeting vim information, e.g. "vdur.10.vim_info.vim:id"
1364         # path_item: dot separated list targeting record information, e.g. "vdur.10"
1365         if ro_vim_item_update:
1366             update_dict = {path_vim_status + "." + k: v for k, v in ro_vim_item_update.items() if k in
1367                            ('vim_id', 'vim_details', 'vim_name', 'vim_status', 'interfaces')}
1368             if path_vim_status.startswith("vdur."):
1369                 # for backward compatibility, add vdur.name apart from vdur.vim_name
1370                 if ro_vim_item_update.get("vim_name"):
1371                     update_dict[path_item + ".name"] = ro_vim_item_update["vim_name"]
1372                 # for backward compatibility, add vdur.vim-id apart from vdur.vim_id
1373                 if ro_vim_item_update.get("vim_id"):
1374                     update_dict[path_item + ".vim-id"] = ro_vim_item_update["vim_id"]
1375                 # update general status
1376                 if ro_vim_item_update.get("vim_status"):
1377                     update_dict[path_item + ".status"] = ro_vim_item_update["vim_status"]
1378             if ro_vim_item_update.get("interfaces"):
1379                 path_interfaces = path_item + ".interfaces"
1380                 for i, iface in enumerate(ro_vim_item_update.get("interfaces")):
1381                     if iface:
1382                         update_dict.update({path_interfaces + ".{}.".format(i) + k: v for k, v in iface.items() if
1383                                             k in ('vlan', 'compute_node', 'pci')})
1384                         # put ip_address and mac_address with ip-address and mac-address
1385                         if iface.get('ip_address'):
1386                             update_dict[path_interfaces + ".{}.".format(i) + "ip-address"] = iface['ip_address']
1387                         if iface.get('mac_address'):
1388                             update_dict[path_interfaces + ".{}.".format(i) + "mac-address"] = iface['mac_address']
1389                         if iface.get("mgmt_vnf_interface") and iface.get("ip_address"):
1390                             update_dict["ip-address"] = iface.get("ip_address").split(";")[0]
1391                         if iface.get("mgmt_vdu_interface") and iface.get("ip_address"):
1392                             update_dict[path_item + ".ip-address"] = iface.get("ip_address").split(";")[0]
1393
1394             self.db.set_one(table, q_filter={"_id": _id}, update_dict=update_dict)
1395         else:
1396             update_dict = {path_item + ".status": "DELETED"}
1397             self.db.set_one(table, q_filter={"_id": _id}, update_dict=update_dict, unset={path_vim_status: None})
1398
1399     def _process_delete_db_tasks(self):
1400         """
1401         Delete task from database because vnfrs or nsrs or both have been deleted
1402         :return: None. Uses and modify self.tasks_to_delete
1403         """
1404         while self.tasks_to_delete:
1405             task = self.tasks_to_delete[0]
1406             vnfrs_deleted = None
1407             nsr_id = task["nsr_id"]
1408             if task["target_record"].startswith("vnfrs:"):
1409                 # check if nsrs is present
1410                 if self.db.get_one("nsrs", {"_id": nsr_id}, fail_on_empty=False):
1411                     vnfrs_deleted = task["target_record"].split(":")[1]
1412             try:
1413                 self.delete_db_tasks(self.db, nsr_id, vnfrs_deleted)
1414             except Exception as e:
1415                 self.logger.error("Error deleting task={}: {}".format(task["task_id"], e))
1416             self.tasks_to_delete.pop(0)
1417
1418     @staticmethod
1419     def delete_db_tasks(db, nsr_id, vnfrs_deleted):
1420         """
1421         Static method because it is called from osm_ng_ro.ns
1422         :param db: instance of database to use
1423         :param nsr_id: affected nsrs id
1424         :param vnfrs_deleted: only tasks with this vnfr id. If None, all affected by nsr_id
1425         :return: None, exception is fails
1426         """
1427         retries = 5
1428         for retry in range(retries):
1429             ro_tasks = db.get_list("ro_tasks", {"tasks.nsr_id": nsr_id})
1430             now = time.time()
1431             conflict = False
1432             for ro_task in ro_tasks:
1433                 db_update = {}
1434                 to_delete_ro_task = True
1435                 for index, task in enumerate(ro_task["tasks"]):
1436                     if not task:
1437                         pass
1438                     elif (not vnfrs_deleted and task["nsr_id"] == nsr_id) or \
1439                             (vnfrs_deleted and task["target_record"].startswith("vnfrs:"+vnfrs_deleted)):
1440                         db_update["tasks.{}".format(index)] = None
1441                     else:
1442                         to_delete_ro_task = False  # used by other nsr, ro_task cannot be deleted
1443                 # delete or update if nobody has changed ro_task meanwhile. Used modified_at for known if changed
1444                 if to_delete_ro_task:
1445                     if not db.del_one("ro_tasks",
1446                                       q_filter={"_id": ro_task["_id"], "modified_at": ro_task["modified_at"]},
1447                                       fail_on_empty=False):
1448                         conflict = True
1449                 elif db_update:
1450                     db_update["modified_at"] = now
1451                     if not db.set_one("ro_tasks",
1452                                       q_filter={"_id": ro_task["_id"], "modified_at": ro_task["modified_at"]},
1453                                       update_dict=db_update,
1454                                       fail_on_empty=False):
1455                         conflict = True
1456             if not conflict:
1457                 return
1458         else:
1459             raise NsWorkerException("Exceeded {} retries".format(retries))
1460
1461     def run(self):
1462         # load database
1463         self.logger.info("Starting")
1464         while True:
1465             # step 1: get commands from queue
1466             try:
1467                 if self.vim_targets:
1468                     task = self.task_queue.get(block=False)
1469                 else:
1470                     if not self.idle:
1471                         self.logger.debug("enters in idle state")
1472                     self.idle = True
1473                     task = self.task_queue.get(block=True)
1474                     self.idle = False
1475
1476                 if task[0] == "terminate":
1477                     break
1478                 elif task[0] == "load_vim":
1479                     self.logger.info("order to load vim {}".format(task[1]))
1480                     self._load_vim(task[1])
1481                 elif task[0] == "unload_vim":
1482                     self.logger.info("order to unload vim {}".format(task[1]))
1483                     self._unload_vim(task[1])
1484                 elif task[0] == "reload_vim":
1485                     self._reload_vim(task[1])
1486                 elif task[0] == "check_vim":
1487                     self.logger.info("order to check vim {}".format(task[1]))
1488                     self._check_vim(task[1])
1489                 continue
1490             except Exception as e:
1491                 if isinstance(e, queue.Empty):
1492                     pass
1493                 else:
1494                     self.logger.critical("Error processing task: {}".format(e), exc_info=True)
1495
1496             # step 2: process pending_tasks, delete not needed tasks
1497             try:
1498                 if self.tasks_to_delete:
1499                     self._process_delete_db_tasks()
1500                 busy = False
1501                 ro_task = self._get_db_task()
1502                 if ro_task:
1503                     self._process_pending_tasks(ro_task)
1504                     busy = True
1505                 if not busy:
1506                     time.sleep(5)
1507             except Exception as e:
1508                 self.logger.critical("Unexpected exception at run: " + str(e), exc_info=True)
1509
1510         self.logger.info("Finishing")