osm_ro/vim_thread.py

   1 # -*- coding: utf-8 -*-
   2
   3 ##
   4 # Copyright 2015 Telefonica Investigacion 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 interacts with a VIM. It processes TASKs sequentially against a single VIM.
  26 The tasks are stored at database in table vim_wim_actions
  27 Several vim_wim_actions can refer to the same element at VIM (flavor, network, ...). This is somethng to avoid if RO
  28 is migrated to a non-relational database as mongo db. Each vim_wim_actions reference a different instance_Xxxxx
  29 In this case "related" colunm contains the same value, to know they refer to the same vim. In case of deletion, it
  30 there is related tasks using this element, it is not deleted, The vim_info needed to delete is transfered to other task
  31
  32 The task content is (M: stored at memory, D: stored at database):
  33     MD  instance_action_id:  reference a global action over an instance-scenario: database instance_actions
  34     MD  task_index:     index number of the task. This together with the previous forms a unique key identifier
  35     MD  datacenter_vim_id:  should contain the uuid of the VIM managed by this thread
  36     MD  vim_id:     id of the vm,net,etc at VIM
  37     MD  item:       database table name, can be instance_vms, instance_nets, TODO: datacenter_flavors, datacenter_images
  38     MD  item_id:    uuid of the referenced entry in the previous table
  39     MD  action:     CREATE, DELETE, FIND
  40     MD  status:     SCHEDULED: action need to be done
  41                     BUILD: not used
  42                     DONE: Done and it must be polled to VIM periodically to see status. ONLY for action=CREATE or FIND
  43                     FAILED: It cannot be created/found/deleted
  44                     FINISHED: similar to DONE, but no refresh is needed anymore. Task is maintained at database but
  45                         it is never processed by any thread
  46                     SUPERSEDED: similar to FINSISHED, but nothing has been done to completed the task.
  47     MD  extra:      text with yaml format at database, dict at memory with:
  48             params:     list with the params to be sent to the VIM for CREATE or FIND. For DELETE the vim_id is taken
  49                         from other related tasks
  50             find:       (only for CREATE tasks) if present it should FIND before creating and use if existing. Contains
  51                         the FIND params
  52             depends_on: list with the 'task_index'es of tasks that must be completed before. e.g. a vm creation depends
  53                         on a net creation
  54                         can contain an int (single index on the same instance-action) or str (compete action ID)
  55             sdn_net_id: used for net.
  56             interfaces: used for VMs. Each key is the uuid of the instance_interfaces entry at database
  57                 iface_id: uuid of intance_interfaces
  58                 sdn_port_id:
  59                 sdn_net_id:
  60                 vim_info
  61             created_items: dictionary with extra elements created that need to be deleted. e.g. ports, volumes,...
  62             created:    False if the VIM element is not created by other actions, and it should not be deleted
  63             vim_status: VIM status of the element. Stored also at database in the instance_XXX
  64             vim_info:   Detailed information of a vm/net from the VIM. Stored at database in the instance_XXX but not at
  65                         vim_wim_actions
  66     M   depends:    dict with task_index(from depends_on) to dependency task
  67     M   params:     same as extra[params]
  68     MD  error_msg:  descriptive text upon an error.Stored also at database instance_XXX
  69     MD  created_at: task creation time. The task of creation must be the oldest
  70     MD  modified_at: next time task need to be processed. For example, for a refresh, it contain next time refresh must
  71                      be done
  72     MD related:     All the tasks over the same VIM element have same "related". Note that other VIMs can contain the
  73                     same value of related, but this thread only process those task of one VIM.  Also related can be the
  74                     same among several NS os isntance-scenarios
  75     MD worker:      Used to lock in case of several thread workers.
  76
  77 """
  78
  79 import threading
  80 import time
  81 import Queue
  82 import logging
  83 import vimconn
  84 import vimconn_openvim
  85 import vimconn_aws
  86 import vimconn_opennebula
  87 import vimconn_openstack
  88 import vimconn_vmware
  89 import yaml
  90 from db_base import db_base_Exception
  91 from lib_osm_openvim.ovim import ovimException
  92 from copy import deepcopy
  93
  94 __author__ = "Alfonso Tierno, Pablo Montes"
  95 __date__ = "$28-Sep-2017 12:07:15$"
  96
  97 vim_module = {
  98     "openvim": vimconn_openvim,
  99     "aws": vimconn_aws,
 100     "opennebula": vimconn_opennebula,
 101     "openstack": vimconn_openstack,
 102     "vmware": vimconn_vmware,
 103 }
 104
 105
 106 def is_task_id(task_id):
 107     return task_id.startswith("TASK-")
 108
 109
 110 class VimThreadException(Exception):
 111     pass
 112
 113
 114 class VimThreadExceptionNotFound(VimThreadException):
 115     pass
 116
 117
 118 class vim_thread(threading.Thread):
 119     REFRESH_BUILD = 5  # 5 seconds
 120     REFRESH_ACTIVE = 60  # 1 minute
 121     REFRESH_ERROR = 600
 122     REFRESH_DELETE = 3600 * 10
 123
 124     def __init__(self, task_lock, name=None, datacenter_name=None, datacenter_tenant_id=None,
 125                  db=None, db_lock=None, ovim=None):
 126         """Init a thread.
 127         Arguments:
 128             'id' number of thead
 129             'name' name of thread
 130             'host','user':  host ip or name to manage and user
 131             'db', 'db_lock': database class and lock to use it in exclusion
 132         """
 133         threading.Thread.__init__(self)
 134         self.vim = None
 135         self.error_status = None
 136         self.datacenter_name = datacenter_name
 137         self.datacenter_tenant_id = datacenter_tenant_id
 138         self.ovim = ovim
 139         if not name:
 140             self.name = vimconn["id"] + "." + vimconn["config"]["datacenter_tenant_id"]
 141         else:
 142             self.name = name
 143         self.vim_persistent_info = {}
 144         self.my_id = self.name[:64]
 145
 146         self.logger = logging.getLogger('openmano.vim.' + self.name)
 147         self.db = db
 148         self.db_lock = db_lock
 149
 150         self.task_lock = task_lock
 151         self.task_queue = Queue.Queue(2000)
 152
 153     def get_vimconnector(self):
 154         try:
 155             from_ = "datacenter_tenants as dt join datacenters as d on dt.datacenter_id=d.uuid"
 156             select_ = ('type', 'd.config as config', 'd.uuid as datacenter_id', 'vim_url', 'vim_url_admin',
 157                        'd.name as datacenter_name', 'dt.uuid as datacenter_tenant_id',
 158                        'dt.vim_tenant_name as vim_tenant_name', 'dt.vim_tenant_id as vim_tenant_id',
 159                        'user', 'passwd', 'dt.config as dt_config')
 160             where_ = {"dt.uuid": self.datacenter_tenant_id}
 161             vims = self.db.get_rows(FROM=from_, SELECT=select_, WHERE=where_)
 162             vim = vims[0]
 163             vim_config = {}
 164             if vim["config"]:
 165                 vim_config.update(yaml.load(vim["config"]))
 166             if vim["dt_config"]:
 167                 vim_config.update(yaml.load(vim["dt_config"]))
 168             vim_config['datacenter_tenant_id'] = vim.get('datacenter_tenant_id')
 169             vim_config['datacenter_id'] = vim.get('datacenter_id')
 170
 171             # get port_mapping
 172             with self.db_lock:
 173                 vim_config["wim_external_ports"] = self.ovim.get_of_port_mappings(
 174                     db_filter={"region": vim_config['datacenter_id'], "pci": None})
 175
 176             self.vim = vim_module[vim["type"]].vimconnector(
 177                 uuid=vim['datacenter_id'], name=vim['datacenter_name'],
 178                 tenant_id=vim['vim_tenant_id'], tenant_name=vim['vim_tenant_name'],
 179                 url=vim['vim_url'], url_admin=vim['vim_url_admin'],
 180                 user=vim['user'], passwd=vim['passwd'],
 181                 config=vim_config, persistent_info=self.vim_persistent_info
 182             )
 183             self.error_status = None
 184         except Exception as e:
 185             self.logger.error("Cannot load vimconnector for vim_account {}: {}".format(self.datacenter_tenant_id, e))
 186             self.vim = None
 187             self.error_status = "Error loading vimconnector: {}".format(e)
 188
 189     def _get_db_task(self):
 190         """
 191         Read actions from database and reload them at memory. Fill self.refresh_list, pending_list, vim_actions
 192         :return: None
 193         """
 194         now = time.time()
 195         try:
 196             database_limit = 20
 197             task_related = None
 198             while True:
 199                 # get 20 (database_limit) entries each time
 200                 vim_actions = self.db.get_rows(FROM="vim_wim_actions",
 201                                                WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
 202                                                       "status": ['SCHEDULED', 'BUILD', 'DONE'],
 203                                                       "worker": [None, self.my_id], "modified_at<=": now
 204                                                       },
 205                                                ORDER_BY=("modified_at", "created_at",),
 206                                                LIMIT=database_limit)
 207                 if not vim_actions:
 208                     return None, None
 209                 # if vim_actions[0]["modified_at"] > now:
 210                 #     return int(vim_actions[0] - now)
 211                 for task in vim_actions:
 212                     # block related task
 213                     if task_related == task["related"]:
 214                         continue  # ignore if a locking has already tried for these task set
 215                     task_related = task["related"]
 216                     # lock ...
 217                     self.db.update_rows("vim_wim_actions", UPDATE={"worker": self.my_id}, modified_time=0,
 218                                         WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
 219                                                "status": ['SCHEDULED', 'BUILD', 'DONE', 'FAILED'],
 220                                                "worker": [None, self.my_id],
 221                                                "related": task_related,
 222                                                "item": task["item"],
 223                                                })
 224                     # ... and read all related and check if locked
 225                     related_tasks = self.db.get_rows(FROM="vim_wim_actions",
 226                                                      WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
 227                                                             "status": ['SCHEDULED', 'BUILD', 'DONE', 'FAILED'],
 228                                                             "related": task_related,
 229                                                             "item": task["item"],
 230                                                             },
 231                                                      ORDER_BY=("created_at",))
 232                     # check that all related tasks have been locked. If not release and try again. It can happen
 233                     # for race conditions if a new related task has been inserted by nfvo in the process
 234                     some_tasks_locked = False
 235                     some_tasks_not_locked = False
 236                     creation_task = None
 237                     for relate_task in related_tasks:
 238                         if relate_task["worker"] != self.my_id:
 239                             some_tasks_not_locked = True
 240                         else:
 241                             some_tasks_locked = True
 242                         if not creation_task and relate_task["action"] in ("CREATE", "FIND"):
 243                             creation_task = relate_task
 244                     if some_tasks_not_locked:
 245                         if some_tasks_locked:  # unlock
 246                             self.db.update_rows("vim_wim_actions", UPDATE={"worker": None}, modified_time=0,
 247                                                 WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
 248                                                        "worker": self.my_id,
 249                                                        "related": task_related,
 250                                                        "item": task["item"],
 251                                                        })
 252                         continue
 253
 254                     # task of creation must be the first in the list of related_task
 255                     assert(related_tasks[0]["action"] in ("CREATE", "FIND"))
 256
 257                     if task["extra"]:
 258                         extra = yaml.load(task["extra"])
 259                     else:
 260                         extra = {}
 261                     task["extra"] = extra
 262                     if extra.get("depends_on"):
 263                         task["depends"] = {}
 264                     if extra.get("params"):
 265                         task["params"] = deepcopy(extra["params"])
 266                     return task, related_tasks
 267         except Exception as e:
 268             self.logger.critical("Unexpected exception at _get_db_task: " + str(e), exc_info=True)
 269             return None, None
 270
 271     def _delete_task(self, task):
 272         """
 273         Determine if this task need to be done or superseded
 274         :return: None
 275         """
 276
 277         def copy_extra_created(copy_to, copy_from):
 278             copy_to["created"] = copy_from["created"]
 279             if copy_from.get("sdn_net_id"):
 280                 copy_to["sdn_net_id"] = copy_from["sdn_net_id"]
 281             if copy_from.get("interfaces"):
 282                 copy_to["interfaces"] = copy_from["interfaces"]
 283             if copy_from.get("created_items"):
 284                 if not copy_to.get("created_items"):
 285                     copy_to["created_items"] = {}
 286                 copy_to["created_items"].update(copy_from["created_items"])
 287
 288         task_create = None
 289         dependency_task = None
 290         deletion_needed = False
 291         if task["status"] == "FAILED":
 292             return   # TODO need to be retry??
 293         try:
 294             # get all related tasks
 295             related_tasks = self.db.get_rows(FROM="vim_wim_actions",
 296                                              WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
 297                                                     "status": ['SCHEDULED', 'BUILD', 'DONE', 'FAILED'],
 298                                                     "action": ["FIND", "CREATE"],
 299                                                     "related": task["related"],
 300                                                     },
 301                                              ORDER_BY=("created_at",),
 302                                              )
 303             for related_task in related_tasks:
 304                 if related_task["item"] == task["item"] and related_task["item_id"] == task["item_id"]:
 305                     task_create = related_task
 306                     # TASK_CREATE
 307                     if related_task["extra"]:
 308                         extra_created = yaml.load(related_task["extra"])
 309                         if extra_created.get("created"):
 310                             deletion_needed = True
 311                         related_task["extra"] = extra_created
 312                 elif not dependency_task:
 313                     dependency_task = related_task
 314                 if task_create and dependency_task:
 315                     break
 316
 317             # mark task_create as FINISHED
 318             self.db.update_rows("vim_wim_actions", UPDATE={"status": "FINISHED"},
 319                                 WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
 320                                        "instance_action_id": task_create["instance_action_id"],
 321                                        "task_index": task_create["task_index"]
 322                                        })
 323             if not deletion_needed:
 324                 return
 325             elif dependency_task:
 326                 # move create information  from task_create to relate_task
 327                 extra_new_created = yaml.load(dependency_task["extra"]) or {}
 328                 extra_new_created["created"] = extra_created["created"]
 329                 copy_extra_created(copy_to=extra_new_created, copy_from=extra_created)
 330
 331                 self.db.update_rows("vim_wim_actions",
 332                                     UPDATE={"extra": yaml.safe_dump(extra_new_created, default_flow_style=True,
 333                                                                     width=256),
 334                                             "vim_id": task_create.get("vim_id")},
 335                                     WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
 336                                            "instance_action_id": dependency_task["instance_action_id"],
 337                                            "task_index": dependency_task["task_index"]
 338                                            })
 339                 return False
 340             else:
 341                 task["vim_id"] = task_create["vim_id"]
 342                 copy_extra_created(copy_to=task["extra"], copy_from=task_create["extra"])
 343                 return True
 344
 345         except Exception as e:
 346             self.logger.critical("Unexpected exception at _delete_task: " + str(e), exc_info=True)
 347
 348     def _refres_vm(self, task):
 349         """Call VIM to get VMs status"""
 350         database_update = None
 351
 352         vim_id = task["vim_id"]
 353         vm_to_refresh_list = [vim_id]
 354         try:
 355             vim_dict = self.vim.refresh_vms_status(vm_to_refresh_list)
 356             vim_info = vim_dict[vim_id]
 357         except vimconn.vimconnException as e:
 358             # Mark all tasks at VIM_ERROR status
 359             self.logger.error("task=several get-VM: vimconnException when trying to refresh vms " + str(e))
 360             vim_info = {"status": "VIM_ERROR", "error_msg": str(e)}
 361
 362         task_id = task["instance_action_id"] + "." + str(task["task_index"])
 363         self.logger.debug("task={} get-VM: vim_vm_id={} result={}".format(task_id, task["vim_id"], vim_info))
 364
 365         # check and update interfaces
 366         task_warning_msg = ""
 367         for interface in vim_info.get("interfaces", ()):
 368             vim_interface_id = interface["vim_interface_id"]
 369             if vim_interface_id not in task["extra"]["interfaces"]:
 370                 self.logger.critical("task={} get-VM: Interface not found {} on task info {}".format(
 371                     task_id, vim_interface_id, task["extra"]["interfaces"]), exc_info=True)
 372                 continue
 373             task_interface = task["extra"]["interfaces"][vim_interface_id]
 374             task_vim_interface = task_interface.get("vim_info")
 375             if task_vim_interface != interface:
 376                 # delete old port
 377                 if task_interface.get("sdn_port_id"):
 378                     try:
 379                         with self.db_lock:
 380                             self.ovim.delete_port(task_interface["sdn_port_id"], idempotent=True)
 381                             task_interface["sdn_port_id"] = None
 382                     except ovimException as e:
 383                         error_text = "ovimException deleting external_port={}: {}".format(
 384                             task_interface["sdn_port_id"], e)
 385                         self.logger.error("task={} get-VM: {}".format(task_id, error_text), exc_info=True)
 386                         task_warning_msg += error_text
 387                         # TODO Set error_msg at instance_nets instead of instance VMs
 388
 389                 # Create SDN port
 390                 sdn_net_id = task_interface.get("sdn_net_id")
 391                 if sdn_net_id and interface.get("compute_node") and interface.get("pci"):
 392                     sdn_port_name = sdn_net_id + "." + task["vim_id"]
 393                     sdn_port_name = sdn_port_name[:63]
 394                     try:
 395                         with self.db_lock:
 396                             sdn_port_id = self.ovim.new_external_port(
 397                                 {"compute_node": interface["compute_node"],
 398                                     "pci": interface["pci"],
 399                                     "vlan": interface.get("vlan"),
 400                                     "net_id": sdn_net_id,
 401                                     "region": self.vim["config"]["datacenter_id"],
 402                                     "name": sdn_port_name,
 403                                     "mac": interface.get("mac_address")})
 404                             task_interface["sdn_port_id"] = sdn_port_id
 405                     except (ovimException, Exception) as e:
 406                         error_text = "ovimException creating new_external_port compute_node={} pci={} vlan={} {}".\
 407                             format(interface["compute_node"], interface["pci"], interface.get("vlan"), e)
 408                         self.logger.error("task={} get-VM: {}".format(task_id, error_text), exc_info=True)
 409                         task_warning_msg += error_text
 410                         # TODO Set error_msg at instance_nets instead of instance VMs
 411
 412                 self.db.update_rows('instance_interfaces',
 413                                     UPDATE={"mac_address": interface.get("mac_address"),
 414                                             "ip_address": interface.get("ip_address"),
 415                                             "vim_interface_id": interface.get("vim_interface_id"),
 416                                             "vim_info": interface.get("vim_info"),
 417                                             "sdn_port_id": task_interface.get("sdn_port_id"),
 418                                             "compute_node": interface.get("compute_node"),
 419                                             "pci": interface.get("pci"),
 420                                             "vlan": interface.get("vlan")},
 421                                     WHERE={'uuid': task_interface["iface_id"]})
 422                 task_interface["vim_info"] = interface
 423
 424         # check and update task and instance_vms database
 425         vim_info_error_msg = None
 426         if vim_info.get("error_msg"):
 427             vim_info_error_msg = self._format_vim_error_msg(vim_info["error_msg"] + task_warning_msg)
 428         elif task_warning_msg:
 429             vim_info_error_msg = self._format_vim_error_msg(task_warning_msg)
 430         task_vim_info = task["extra"].get("vim_info")
 431         task_error_msg = task.get("error_msg")
 432         task_vim_status = task["extra"].get("vim_status")
 433         if task_vim_status != vim_info["status"] or task_error_msg != vim_info_error_msg or \
 434                 (vim_info.get("vim_info") and task_vim_info != vim_info["vim_info"]):
 435             database_update = {"status": vim_info["status"], "error_msg": vim_info_error_msg}
 436             if vim_info.get("vim_info"):
 437                 database_update["vim_info"] = vim_info["vim_info"]
 438
 439             task["extra"]["vim_status"] = vim_info["status"]
 440             task["error_msg"] = vim_info_error_msg
 441             if vim_info.get("vim_info"):
 442                 task["extra"]["vim_info"] = vim_info["vim_info"]
 443
 444         return database_update
 445
 446     def _refres_net(self, task):
 447         """Call VIM to get network status"""
 448         database_update = None
 449
 450         vim_id = task["vim_id"]
 451         net_to_refresh_list = [vim_id]
 452         try:
 453             vim_dict = self.vim.refresh_nets_status(net_to_refresh_list)
 454             vim_info = vim_dict[vim_id]
 455         except vimconn.vimconnException as e:
 456             # Mark all tasks at VIM_ERROR status
 457             self.logger.error("task=several get-net: vimconnException when trying to refresh nets " + str(e))
 458             vim_info = {"status": "VIM_ERROR", "error_msg": str(e)}
 459
 460         task_id = task["instance_action_id"] + "." + str(task["task_index"])
 461         self.logger.debug("task={} get-net: vim_net_id={} result={}".format(task_id, task["vim_id"], vim_info))
 462
 463         task_vim_info = task["extra"].get("vim_info")
 464         task_vim_status = task["extra"].get("vim_status")
 465         task_error_msg = task.get("error_msg")
 466         task_sdn_net_id = task["extra"].get("sdn_net_id")
 467
 468         vim_info_status = vim_info["status"]
 469         vim_info_error_msg = vim_info.get("error_msg")
 470         # get ovim status
 471         if task_sdn_net_id:
 472             try:
 473                 with self.db_lock:
 474                     sdn_net = self.ovim.show_network(task_sdn_net_id)
 475             except (ovimException, Exception) as e:
 476                 text_error = "ovimException getting network snd_net_id={}: {}".format(task_sdn_net_id, e)
 477                 self.logger.error("task={} get-net: {}".format(task_id, text_error), exc_info=True)
 478                 sdn_net = {"status": "ERROR", "last_error": text_error}
 479             if sdn_net["status"] == "ERROR":
 480                 if not vim_info_error_msg:
 481                     vim_info_error_msg = str(sdn_net.get("last_error"))
 482                 else:
 483                     vim_info_error_msg = "VIM_ERROR: {} && SDN_ERROR: {}".format(
 484                         self._format_vim_error_msg(vim_info_error_msg, 1024 // 2 - 14),
 485                         self._format_vim_error_msg(sdn_net["last_error"], 1024 // 2 - 14))
 486                 vim_info_status = "ERROR"
 487             elif sdn_net["status"] == "BUILD":
 488                 if vim_info_status == "ACTIVE":
 489                     vim_info_status = "BUILD"
 490
 491         # update database
 492         if vim_info_error_msg:
 493             vim_info_error_msg = self._format_vim_error_msg(vim_info_error_msg)
 494         if task_vim_status != vim_info_status or task_error_msg != vim_info_error_msg or \
 495                 (vim_info.get("vim_info") and task_vim_info != vim_info["vim_info"]):
 496             task["extra"]["vim_status"] = vim_info_status
 497             task["error_msg"] = vim_info_error_msg
 498             if vim_info.get("vim_info"):
 499                 task["extra"]["vim_info"] = vim_info["vim_info"]
 500             database_update = {"status": vim_info_status, "error_msg": vim_info_error_msg}
 501             if vim_info.get("vim_info"):
 502                 database_update["vim_info"] = vim_info["vim_info"]
 503         return database_update
 504
 505     def _proccess_pending_tasks(self, task, related_tasks):
 506         old_task_status = task["status"]
 507         create_or_find = False   # if as result of processing this task something is created or found
 508         next_refresh = 0
 509
 510         try:
 511             if task["status"] == "SCHEDULED":
 512                 # check if tasks that this depends on have been completed
 513                 dependency_not_completed = False
 514                 dependency_modified_at = 0
 515                 for task_index in task["extra"].get("depends_on", ()):
 516                     task_dependency = self._look_for_task(task["instance_action_id"], task_index)
 517                     if not task_dependency:
 518                         raise VimThreadException(
 519                             "Cannot get depending net task trying to get depending task {}.{}".format(
 520                                 task["instance_action_id"], task_index))
 521                     # task["depends"]["TASK-" + str(task_index)] = task_dependency #it references another object,so
 522                     # database must be look again
 523                     if task_dependency["status"] == "SCHEDULED":
 524                         dependency_not_completed = True
 525                         dependency_modified_at = task_dependency["modified_at"]
 526                         break
 527                     elif task_dependency["status"] == "FAILED":
 528                         raise VimThreadException(
 529                             "Cannot {} {}, (task {}.{}) because depends on failed {}.{}, (task{}.{}): {}".format(
 530                                 task["action"], task["item"],
 531                                 task["instance_action_id"], task["task_index"],
 532                                 task_dependency["instance_action_id"], task_dependency["task_index"],
 533                                 task_dependency["action"], task_dependency["item"], task_dependency.get("error_msg")))
 534
 535                     task["depends"]["TASK-"+str(task_index)] = task_dependency
 536                     task["depends"]["TASK-{}.{}".format(task["instance_action_id"], task_index)] = task_dependency
 537                 if dependency_not_completed:
 538                     # Move this task to the time dependency is going to be modified plus 10 seconds.
 539                     self.db.update_rows("vim_wim_actions", modified_time=dependency_modified_at + 10,
 540                                         UPDATE={"worker": None},
 541                                         WHERE={"datacenter_vim_id": self.datacenter_tenant_id, "worker": self.my_id,
 542                                                "related": task["related"],
 543                                                })
 544                     # task["extra"]["tries"] = task["extra"].get("tries", 0) + 1
 545                     # if task["extra"]["tries"] > 3:
 546                     #     raise VimThreadException(
 547                     #         "Cannot {} {}, (task {}.{}) because timeout waiting to complete {} {}, "
 548                     #         "(task {}.{})".format(task["action"], task["item"],
 549                     #                               task["instance_action_id"], task["task_index"],
 550                     #                               task_dependency["instance_action_id"], task_dependency["task_index"]
 551                     #                               task_dependency["action"], task_dependency["item"]))
 552                     return
 553
 554             database_update = None
 555             if task["action"] == "DELETE":
 556                 deleted_needed = self._delete_task(task)
 557                 if not deleted_needed:
 558                     task["status"] = "SUPERSEDED"  # with FINISHED instead of DONE it will not be refreshing
 559                     task["error_msg"] = None
 560
 561             if task["status"] == "SUPERSEDED":
 562                 # not needed to do anything but update database with the new status
 563                 database_update = None
 564             elif not self.vim:
 565                 task["status"] = "FAILED"
 566                 task["error_msg"] = self.error_status
 567                 database_update = {"status": "VIM_ERROR", "error_msg": task["error_msg"]}
 568             elif task["item_id"] != related_tasks[0]["item_id"] and task["action"] in ("FIND", "CREATE"):
 569                 # Do nothing, just copy values from one to another and updata database
 570                 task["status"] = related_tasks[0]["status"]
 571                 task["error_msg"] = related_tasks[0]["error_msg"]
 572                 task["vim_id"] = related_tasks[0]["vim_id"]
 573                 extra = yaml.load(related_tasks[0]["extra"])
 574                 task["extra"]["vim_status"] = extra["vim_status"]
 575                 next_refresh = related_tasks[0]["modified_at"] + 0.001
 576                 database_update = {"status": task["extra"].get("vim_status", "VIM_ERROR"),
 577                                    "error_msg": task["error_msg"]}
 578                 if task["item"] == 'instance_vms':
 579                     database_update["vim_vm_id"] = task["vim_id"]
 580                 elif task["item"] == 'instance_nets':
 581                     database_update["vim_net_id"] = task["vim_id"]
 582             elif task["item"] == 'instance_vms':
 583                 if task["status"] in ('BUILD', 'DONE') and task["action"] in ("FIND", "CREATE"):
 584                     database_update = self._refres_vm(task)
 585                     create_or_find = True
 586                 elif task["action"] == "CREATE":
 587                     create_or_find = True
 588                     database_update = self.new_vm(task)
 589                 elif task["action"] == "DELETE":
 590                     self.del_vm(task)
 591                 else:
 592                     raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
 593             elif task["item"] == 'instance_nets':
 594                 if task["status"] in ('BUILD', 'DONE') and task["action"] in ("FIND", "CREATE"):
 595                     database_update = self._refres_net(task)
 596                     create_or_find = True
 597                 elif task["action"] == "CREATE":
 598                     create_or_find = True
 599                     database_update = self.new_net(task)
 600                 elif task["action"] == "DELETE":
 601                     self.del_net(task)
 602                 elif task["action"] == "FIND":
 603                     database_update = self.get_net(task)
 604                 else:
 605                     raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
 606             elif task["item"] == 'instance_sfis':
 607                 if task["action"] == "CREATE":
 608                     create_or_find = True
 609                     database_update = self.new_sfi(task)
 610                 elif task["action"] == "DELETE":
 611                     self.del_sfi(task)
 612                 else:
 613                     raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
 614             elif task["item"] == 'instance_sfs':
 615                 if task["action"] == "CREATE":
 616                     create_or_find = True
 617                     database_update = self.new_sf(task)
 618                 elif task["action"] == "DELETE":
 619                     self.del_sf(task)
 620                 else:
 621                     raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
 622             elif task["item"] == 'instance_classifications':
 623                 if task["action"] == "CREATE":
 624                     create_or_find = True
 625                     database_update = self.new_classification(task)
 626                 elif task["action"] == "DELETE":
 627                     self.del_classification(task)
 628                 else:
 629                     raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
 630             elif task["item"] == 'instance_sfps':
 631                 if task["action"] == "CREATE":
 632                     create_or_find = True
 633                     database_update = self.new_sfp(task)
 634                 elif task["action"] == "DELETE":
 635                     self.del_sfp(task)
 636                 else:
 637                     raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
 638             else:
 639                 raise vimconn.vimconnException(self.name + "unknown task item {}".format(task["item"]))
 640                 # TODO
 641         except VimThreadException as e:
 642             task["error_msg"] = str(e)
 643             task["status"] = "FAILED"
 644             database_update = {"status": "VIM_ERROR", "error_msg": task["error_msg"]}
 645             if task["item"] == 'instance_vms':
 646                 database_update["vim_vm_id"] = None
 647             elif task["item"] == 'instance_nets':
 648                 database_update["vim_net_id"] = None
 649
 650         task_id = task["instance_action_id"] + "." + str(task["task_index"])
 651         self.logger.debug("task={} item={} action={} result={}:'{}' params={}".format(
 652             task_id, task["item"], task["action"], task["status"],
 653             task["vim_id"] if task["status"] == "DONE" else task.get("error_msg"), task["params"]))
 654         try:
 655             if not next_refresh:
 656                 if task["status"] == "DONE":
 657                     next_refresh = time.time()
 658                     if task["extra"].get("vim_status") == "BUILD":
 659                         next_refresh += self.REFRESH_BUILD
 660                     elif task["extra"].get("vim_status") in ("ERROR", "VIM_ERROR"):
 661                         next_refresh += self.REFRESH_ERROR
 662                     elif task["extra"].get("vim_status") == "DELETED":
 663                         next_refresh += self.REFRESH_DELETE
 664                     else:
 665                         next_refresh += self.REFRESH_ACTIVE
 666                 elif task["status"] == "FAILED":
 667                     next_refresh = time.time() + self.REFRESH_DELETE
 668
 669             if create_or_find:
 670                 # modify all related task with action FIND/CREATED non SCHEDULED
 671                 self.db.update_rows(
 672                     table="vim_wim_actions", modified_time=next_refresh + 0.001,
 673                     UPDATE={"status": task["status"], "vim_id": task.get("vim_id"),
 674                             "error_msg": task["error_msg"],
 675                             },
 676
 677                     WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
 678                            "worker": self.my_id,
 679                            "action": ["FIND", "CREATE"],
 680                            "related": task["related"],
 681                            "status<>": "SCHEDULED",
 682                            })
 683             # modify own task
 684             self.db.update_rows(
 685                 table="vim_wim_actions", modified_time=next_refresh,
 686                 UPDATE={"status": task["status"], "vim_id": task.get("vim_id"),
 687                         "error_msg": task["error_msg"],
 688                         "extra": yaml.safe_dump(task["extra"], default_flow_style=True, width=256)},
 689                 WHERE={"instance_action_id": task["instance_action_id"], "task_index": task["task_index"]})
 690             # Unlock tasks
 691             self.db.update_rows(
 692                 table="vim_wim_actions", modified_time=0,
 693                 UPDATE={"worker": None},
 694                 WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
 695                        "worker": self.my_id,
 696                        "related": task["related"],
 697                        })
 698
 699             # Update table instance_actions
 700             if old_task_status == "SCHEDULED" and task["status"] != old_task_status:
 701                 self.db.update_rows(
 702                     table="instance_actions",
 703                     UPDATE={("number_failed" if task["status"] == "FAILED" else "number_done"): {"INCREMENT": 1}},
 704                     WHERE={"uuid": task["instance_action_id"]})
 705             if database_update:
 706                 self.db.update_rows(table=task["item"],
 707                                     UPDATE=database_update,
 708                                     WHERE={"related": task["related"]})
 709         except db_base_Exception as e:
 710             self.logger.error("task={} Error updating database {}".format(task_id, e), exc_info=True)
 711
 712     def insert_task(self, task):
 713         try:
 714             self.task_queue.put(task, False)
 715             return None
 716         except Queue.Full:
 717             raise vimconn.vimconnException(self.name + ": timeout inserting a task")
 718
 719     def del_task(self, task):
 720         with self.task_lock:
 721             if task["status"] == "SCHEDULED":
 722                 task["status"] = "SUPERSEDED"
 723                 return True
 724             else:  # task["status"] == "processing"
 725                 self.task_lock.release()
 726                 return False
 727
 728     def run(self):
 729         self.logger.debug("Starting")
 730         while True:
 731             self.get_vimconnector()
 732             self.logger.debug("Vimconnector loaded")
 733             reload_thread = False
 734
 735             while True:
 736                 try:
 737                     while not self.task_queue.empty():
 738                         task = self.task_queue.get()
 739                         if isinstance(task, list):
 740                             pass
 741                         elif isinstance(task, str):
 742                             if task == 'exit':
 743                                 return 0
 744                             elif task == 'reload':
 745                                 reload_thread = True
 746                                 break
 747                         self.task_queue.task_done()
 748                     if reload_thread:
 749                         break
 750
 751                     task, related_tasks = self._get_db_task()
 752                     if task:
 753                         self._proccess_pending_tasks(task, related_tasks)
 754                     else:
 755                         time.sleep(5)
 756
 757                 except Exception as e:
 758                     self.logger.critical("Unexpected exception at run: " + str(e), exc_info=True)
 759
 760         self.logger.debug("Finishing")
 761
 762     def _look_for_task(self, instance_action_id, task_id):
 763         """
 764         Look for a concrete task at vim_actions database table
 765         :param instance_action_id: The instance_action_id
 766         :param task_id: Can have several formats:
 767             <task index>: integer
 768             TASK-<task index> :backward compatibility,
 769             [TASK-]<instance_action_id>.<task index>: this instance_action_id overrides the one in the parameter
 770         :return: Task dictionary or None if not found
 771         """
 772         if isinstance(task_id, int):
 773             task_index = task_id
 774         else:
 775             if task_id.startswith("TASK-"):
 776                 task_id = task_id[5:]
 777             ins_action_id, _, task_index = task_id.rpartition(".")
 778             if ins_action_id:
 779                 instance_action_id = ins_action_id
 780
 781         tasks = self.db.get_rows(FROM="vim_wim_actions", WHERE={"instance_action_id": instance_action_id,
 782                                                                 "task_index": task_index})
 783         if not tasks:
 784             return None
 785         task = tasks[0]
 786         task["params"] = None
 787         task["depends"] = {}
 788         if task["extra"]:
 789             extra = yaml.load(task["extra"])
 790             task["extra"] = extra
 791             task["params"] = extra.get("params")
 792         else:
 793             task["extra"] = {}
 794         return task
 795
 796     @staticmethod
 797     def _format_vim_error_msg(error_text, max_length=1024):
 798         if error_text and len(error_text) >= max_length:
 799             return error_text[:max_length // 2 - 3] + " ... " + error_text[-max_length // 2 + 3:]
 800         return error_text
 801
 802     def new_vm(self, task):
 803         task_id = task["instance_action_id"] + "." + str(task["task_index"])
 804         try:
 805             params = task["params"]
 806             depends = task.get("depends")
 807             net_list = params[5]
 808             for net in net_list:
 809                 if "net_id" in net and is_task_id(net["net_id"]):  # change task_id into network_id
 810                     network_id = task["depends"][net["net_id"]].get("vim_id")
 811                     if not network_id:
 812                         raise VimThreadException(
 813                             "Cannot create VM because depends on a network not created or found: " +
 814                             str(depends[net["net_id"]]["error_msg"]))
 815                     net["net_id"] = network_id
 816             params_copy = deepcopy(params)
 817             vim_vm_id, created_items = self.vim.new_vminstance(*params_copy)
 818
 819             # fill task_interfaces. Look for snd_net_id at database for each interface
 820             task_interfaces = {}
 821             for iface in params_copy[5]:
 822                 task_interfaces[iface["vim_id"]] = {"iface_id": iface["uuid"]}
 823                 result = self.db.get_rows(
 824                     SELECT=('sdn_net_id', 'interface_id'),
 825                     FROM='instance_nets as ine join instance_interfaces as ii on ii.instance_net_id=ine.uuid',
 826                     WHERE={'ii.uuid': iface["uuid"]})
 827                 if result:
 828                     task_interfaces[iface["vim_id"]]["sdn_net_id"] = result[0]['sdn_net_id']
 829                     task_interfaces[iface["vim_id"]]["interface_id"] = result[0]['interface_id']
 830                 else:
 831                     self.logger.critical("task={} new-VM: instance_nets uuid={} not found at DB".format(task_id,
 832                                                                                                         iface["uuid"]),
 833                                          exc_info=True)
 834
 835             task["vim_info"] = {}
 836             task["extra"]["interfaces"] = task_interfaces
 837             task["extra"]["created"] = True
 838             task["extra"]["created_items"] = created_items
 839             task["extra"]["vim_status"] = "BUILD"
 840             task["error_msg"] = None
 841             task["status"] = "DONE"
 842             task["vim_id"] = vim_vm_id
 843             instance_element_update = {"status": "BUILD", "vim_vm_id": vim_vm_id, "error_msg": None}
 844             return instance_element_update
 845
 846         except (vimconn.vimconnException, VimThreadException) as e:
 847             self.logger.error("task={} new-VM: {}".format(task_id, e))
 848             error_text = self._format_vim_error_msg(str(e))
 849             task["error_msg"] = error_text
 850             task["status"] = "FAILED"
 851             task["vim_id"] = None
 852             instance_element_update = {"status": "VIM_ERROR", "vim_vm_id": None, "error_msg": error_text}
 853             return instance_element_update
 854
 855     def del_vm(self, task):
 856         task_id = task["instance_action_id"] + "." + str(task["task_index"])
 857         vm_vim_id = task["vim_id"]
 858         interfaces = task["extra"].get("interfaces", ())
 859         try:
 860             for iface in interfaces.values():
 861                 if iface.get("sdn_port_id"):
 862                     try:
 863                         with self.db_lock:
 864                             self.ovim.delete_port(iface["sdn_port_id"], idempotent=True)
 865                     except ovimException as e:
 866                         self.logger.error("task={} del-VM: ovimException when deleting external_port={}: {} ".format(
 867                             task_id, iface["sdn_port_id"], e), exc_info=True)
 868                         # TODO Set error_msg at instance_nets
 869
 870             self.vim.delete_vminstance(vm_vim_id, task["extra"].get("created_items"))
 871             task["status"] = "FINISHED"  # with FINISHED instead of DONE it will not be refreshing
 872             task["error_msg"] = None
 873             return None
 874
 875         except vimconn.vimconnException as e:
 876             task["error_msg"] = self._format_vim_error_msg(str(e))
 877             if isinstance(e, vimconn.vimconnNotFoundException):
 878                 # If not found mark as Done and fill error_msg
 879                 task["status"] = "FINISHED"  # with FINISHED instead of DONE it will not be refreshing
 880                 return None
 881             task["status"] = "FAILED"
 882             return None
 883
 884     def _get_net_internal(self, task, filter_param):
 885         """
 886         Common code for get_net and new_net. It looks for a network on VIM with the filter_params
 887         :param task: task for this find or find-or-create action
 888         :param filter_param: parameters to send to the vimconnector
 889         :return: a dict with the content to update the instance_nets database table. Raises an exception on error, or
 890             when network is not found or found more than one
 891         """
 892         vim_nets = self.vim.get_network_list(filter_param)
 893         if not vim_nets:
 894             raise VimThreadExceptionNotFound("Network not found with this criteria: '{}'".format(filter_param))
 895         elif len(vim_nets) > 1:
 896             raise VimThreadException("More than one network found with this criteria: '{}'".format(filter_param))
 897         vim_net_id = vim_nets[0]["id"]
 898
 899         # Discover if this network is managed by a sdn controller
 900         sdn_net_id = None
 901         result = self.db.get_rows(SELECT=('sdn_net_id',), FROM='instance_nets',
 902                                   WHERE={'vim_net_id': vim_net_id, 'datacenter_tenant_id': self.datacenter_tenant_id},
 903                                   ORDER="instance_scenario_id")
 904         if result:
 905             sdn_net_id = result[0]['sdn_net_id']
 906
 907         task["status"] = "DONE"
 908         task["extra"]["vim_info"] = {}
 909         task["extra"]["created"] = False
 910         task["extra"]["vim_status"] = "BUILD"
 911         task["extra"]["sdn_net_id"] = sdn_net_id
 912         task["error_msg"] = None
 913         task["vim_id"] = vim_net_id
 914         instance_element_update = {"vim_net_id": vim_net_id, "created": False, "status": "BUILD",
 915                                    "error_msg": None, "sdn_net_id": sdn_net_id}
 916         return instance_element_update
 917
 918     def get_net(self, task):
 919         task_id = task["instance_action_id"] + "." + str(task["task_index"])
 920         try:
 921             params = task["params"]
 922             filter_param = params[0]
 923             instance_element_update = self._get_net_internal(task, filter_param)
 924             return instance_element_update
 925
 926         except (vimconn.vimconnException, VimThreadException) as e:
 927             self.logger.error("task={} get-net: {}".format(task_id, e))
 928             task["status"] = "FAILED"
 929             task["vim_id"] = None
 930             task["error_msg"] = self._format_vim_error_msg(str(e))
 931             instance_element_update = {"vim_net_id": None, "status": "VIM_ERROR",
 932                                        "error_msg": task["error_msg"]}
 933             return instance_element_update
 934
 935     def new_net(self, task):
 936         vim_net_id = None
 937         sdn_net_id = None
 938         task_id = task["instance_action_id"] + "." + str(task["task_index"])
 939         action_text = ""
 940         try:
 941             # FIND
 942             if task["extra"].get("find"):
 943                 action_text = "finding"
 944                 filter_param = task["extra"]["find"][0]
 945                 try:
 946                     instance_element_update = self._get_net_internal(task, filter_param)
 947                     return instance_element_update
 948                 except VimThreadExceptionNotFound:
 949                     pass
 950             # CREATE
 951             params = task["params"]
 952             action_text = "creating VIM"
 953             vim_net_id, created_items = self.vim.new_network(*params[0:3])
 954
 955             net_name = params[0]
 956             net_type = params[1]
 957             wim_account_name = None
 958             if len(params) >= 4:
 959                 wim_account_name = params[3]
 960
 961             sdn_controller = self.vim.config.get('sdn-controller')
 962             if sdn_controller and (net_type == "data" or net_type == "ptp"):
 963                 network = {"name": net_name, "type": net_type, "region": self.vim["config"]["datacenter_id"]}
 964
 965                 vim_net = self.vim.get_network(vim_net_id)
 966                 if vim_net.get('encapsulation') != 'vlan':
 967                     raise vimconn.vimconnException(
 968                         "net '{}' defined as type '{}' has not vlan encapsulation '{}'".format(
 969                             net_name, net_type, vim_net['encapsulation']))
 970                 network["vlan"] = vim_net.get('segmentation_id')
 971                 action_text = "creating SDN"
 972                 with self.db_lock:
 973                     sdn_net_id = self.ovim.new_network(network)
 974
 975                 if wim_account_name and self.vim.config["wim_external_ports"]:
 976                     # add external port to connect WIM. Try with compute node __WIM:wim_name and __WIM
 977                     action_text = "attaching external port to ovim network"
 978                     sdn_port_name = sdn_net_id + "." + task["vim_id"]
 979                     sdn_port_name = sdn_port_name[:63]
 980                     sdn_port_data = {
 981                         "compute_node": "__WIM:" + wim_account_name[0:58],
 982                         "pci": None,
 983                         "vlan": network["vlan"],
 984                         "net_id": sdn_net_id,
 985                         "region": self.vim["config"]["datacenter_id"],
 986                         "name": sdn_port_name,
 987                     }
 988                     try:
 989                         with self.db_lock:
 990                             sdn_external_port_id = self.ovim.new_external_port(sdn_port_data)
 991                     except ovimException:
 992                         sdn_port_data["compute_node"] = "__WIM"
 993                         with self.db_lock:
 994                             sdn_external_port_id = self.ovim.new_external_port(sdn_port_data)
 995                     self.logger.debug("Added sdn_external_port {} to sdn_network {}".format(sdn_external_port_id,
 996                                                                                             sdn_net_id))
 997             task["status"] = "DONE"
 998             task["extra"]["vim_info"] = {}
 999             task["extra"]["sdn_net_id"] = sdn_net_id
1000             task["extra"]["vim_status"] = "BUILD"
1001             task["extra"]["created"] = True
1002             task["extra"]["created_items"] = created_items
1003             task["error_msg"] = None
1004             task["vim_id"] = vim_net_id
1005             instance_element_update = {"vim_net_id": vim_net_id, "sdn_net_id": sdn_net_id, "status": "BUILD",
1006                                        "created": True, "error_msg": None}
1007             return instance_element_update
1008         except (vimconn.vimconnException, ovimException) as e:
1009             self.logger.error("task={} new-net: Error {}: {}".format(task_id, action_text, e))
1010             task["status"] = "FAILED"
1011             task["vim_id"] = vim_net_id
1012             task["error_msg"] = self._format_vim_error_msg(str(e))
1013             task["extra"]["sdn_net_id"] = sdn_net_id
1014             instance_element_update = {"vim_net_id": vim_net_id, "sdn_net_id": sdn_net_id, "status": "VIM_ERROR",
1015                                        "error_msg": task["error_msg"]}
1016             return instance_element_update
1017
1018     def del_net(self, task):
1019         net_vim_id = task["vim_id"]
1020         sdn_net_id = task["extra"].get("sdn_net_id")
1021         try:
1022             if sdn_net_id:
1023                 # Delete any attached port to this sdn network. There can be ports associated to this network in case
1024                 # it was manually done using 'openmano vim-net-sdn-attach'
1025                 with self.db_lock:
1026                     port_list = self.ovim.get_ports(columns={'uuid'},
1027                                                     filter={'name': 'external_port', 'net_id': sdn_net_id})
1028                     for port in port_list:
1029                         self.ovim.delete_port(port['uuid'], idempotent=True)
1030                     self.ovim.delete_network(sdn_net_id, idempotent=True)
1031             if net_vim_id:
1032                 self.vim.delete_network(net_vim_id, task["extra"].get("created_items"))
1033             task["status"] = "FINISHED"  # with FINISHED instead of DONE it will not be refreshing
1034             task["error_msg"] = None
1035             return None
1036         except ovimException as e:
1037             task["error_msg"] = self._format_vim_error_msg("ovimException obtaining and deleting external "
1038                                                            "ports for net {}: {}".format(sdn_net_id, str(e)))
1039         except vimconn.vimconnException as e:
1040             task["error_msg"] = self._format_vim_error_msg(str(e))
1041             if isinstance(e, vimconn.vimconnNotFoundException):
1042                 # If not found mark as Done and fill error_msg
1043                 task["status"] = "FINISHED"  # with FINISHED instead of DONE it will not be refreshing
1044                 return None
1045         task["status"] = "FAILED"
1046         return None
1047
1048     # Service Function Instances
1049     def new_sfi(self, task):
1050         vim_sfi_id = None
1051         try:
1052             # Waits for interfaces to be ready (avoids failure)
1053             time.sleep(1)
1054             dep_id = "TASK-" + str(task["extra"]["depends_on"][0])
1055             task_id = task["instance_action_id"] + "." + str(task["task_index"])
1056             error_text = ""
1057             interfaces = task["depends"][dep_id]["extra"].get("interfaces")
1058
1059             ingress_interface_id = task.get("extra").get("params").get("ingress_interface_id")
1060             egress_interface_id = task.get("extra").get("params").get("egress_interface_id")
1061             ingress_vim_interface_id = None
1062             egress_vim_interface_id = None
1063             for vim_interface, interface_data in interfaces.iteritems():
1064                 if interface_data.get("interface_id") == ingress_interface_id:
1065                     ingress_vim_interface_id = vim_interface
1066                     break
1067             if ingress_interface_id != egress_interface_id:
1068                 for vim_interface, interface_data in interfaces.iteritems():
1069                     if interface_data.get("interface_id") == egress_interface_id:
1070                         egress_vim_interface_id = vim_interface
1071                         break
1072             else:
1073                 egress_vim_interface_id = ingress_vim_interface_id
1074             if not ingress_vim_interface_id or not egress_vim_interface_id:
1075                 error_text = "Error creating Service Function Instance, Ingress: {}, Egress: {}".format(
1076                     ingress_vim_interface_id, egress_vim_interface_id)
1077                 self.logger.error(error_text)
1078                 task["error_msg"] = error_text
1079                 task["status"] = "FAILED"
1080                 task["vim_id"] = None
1081                 return None
1082             # At the moment, every port associated with the VM will be used both as ingress and egress ports.
1083             # Bear in mind that different VIM connectors might support SFI differently. In the case of OpenStack,
1084             # only the first ingress and first egress ports will be used to create the SFI (Port Pair).
1085             ingress_port_id_list = [ingress_vim_interface_id]
1086             egress_port_id_list = [egress_vim_interface_id]
1087             name = "sfi-%s" % task["item_id"][:8]
1088             # By default no form of IETF SFC Encapsulation will be used
1089             vim_sfi_id = self.vim.new_sfi(name, ingress_port_id_list, egress_port_id_list, sfc_encap=False)
1090
1091             task["extra"]["created"] = True
1092             task["extra"]["vim_status"] = "ACTIVE"
1093             task["error_msg"] = None
1094             task["status"] = "FINISHED"  # with FINISHED instead of DONE it will not be refreshing
1095             task["vim_id"] = vim_sfi_id
1096             instance_element_update = {"status": "ACTIVE", "vim_sfi_id": vim_sfi_id, "error_msg": None}
1097             return instance_element_update
1098
1099         except (vimconn.vimconnException, VimThreadException) as e:
1100             self.logger.error("Error creating Service Function Instance, task=%s: %s", task_id, str(e))
1101             error_text = self._format_vim_error_msg(str(e))
1102             task["error_msg"] = error_text
1103             task["status"] = "FAILED"
1104             task["vim_id"] = None
1105             instance_element_update = {"status": "VIM_ERROR", "vim_sfi_id": None, "error_msg": error_text}
1106             return instance_element_update
1107
1108     def del_sfi(self, task):
1109         sfi_vim_id = task["vim_id"]
1110         try:
1111             self.vim.delete_sfi(sfi_vim_id)
1112             task["status"] = "FINISHED"  # with FINISHED instead of DONE it will not be refreshing
1113             task["error_msg"] = None
1114             return None
1115
1116         except vimconn.vimconnException as e:
1117             task["error_msg"] = self._format_vim_error_msg(str(e))
1118             if isinstance(e, vimconn.vimconnNotFoundException):
1119                 # If not found mark as Done and fill error_msg
1120                 task["status"] = "FINISHED"  # with FINISHED instead of DONE it will not be refreshing
1121                 return None
1122             task["status"] = "FAILED"
1123             return None
1124
1125     def new_sf(self, task):
1126         vim_sf_id = None
1127         try:
1128             task_id = task["instance_action_id"] + "." + str(task["task_index"])
1129             error_text = ""
1130             depending_tasks = ["TASK-" + str(dep_id) for dep_id in task["extra"]["depends_on"]]
1131             # sfis = task.get("depends").values()[0].get("extra").get("params")[5]
1132             sfis = [task.get("depends").get(dep_task) for dep_task in depending_tasks]
1133             sfi_id_list = []
1134             for sfi in sfis:
1135                 sfi_id_list.append(sfi.get("vim_id"))
1136             name = "sf-%s" % task["item_id"][:8]
1137             # By default no form of IETF SFC Encapsulation will be used
1138             vim_sf_id = self.vim.new_sf(name, sfi_id_list, sfc_encap=False)
1139
1140             task["extra"]["created"] = True
1141             task["extra"]["vim_status"] = "ACTIVE"
1142             task["error_msg"] = None
1143             task["status"] = "FINISHED"  # with FINISHED instead of DONE it will not be refreshing
1144             task["vim_id"] = vim_sf_id
1145             instance_element_update = {"status": "ACTIVE", "vim_sf_id": vim_sf_id, "error_msg": None}
1146             return instance_element_update
1147
1148         except (vimconn.vimconnException, VimThreadException) as e:
1149             self.logger.error("Error creating Service Function, task=%s: %s", task_id, str(e))
1150             error_text = self._format_vim_error_msg(str(e))
1151             task["error_msg"] = error_text
1152             task["status"] = "FAILED"
1153             task["vim_id"] = None
1154             instance_element_update = {"status": "VIM_ERROR", "vim_sf_id": None, "error_msg": error_text}
1155             return instance_element_update
1156
1157     def del_sf(self, task):
1158         sf_vim_id = task["vim_id"]
1159         try:
1160             self.vim.delete_sf(sf_vim_id)
1161             task["status"] = "FINISHED"  # with FINISHED instead of DONE it will not be refreshing
1162             task["error_msg"] = None
1163             return None
1164
1165         except vimconn.vimconnException as e:
1166             task["error_msg"] = self._format_vim_error_msg(str(e))
1167             if isinstance(e, vimconn.vimconnNotFoundException):
1168                 # If not found mark as Done and fill error_msg
1169                 task["status"] = "FINISHED"  # with FINISHED instead of DONE it will not be refreshing
1170                 return None
1171             task["status"] = "FAILED"
1172             return None
1173
1174     def new_classification(self, task):
1175         vim_classification_id = None
1176         try:
1177             params = task["params"]
1178             task_id = task["instance_action_id"] + "." + str(task["task_index"])
1179             dep_id = "TASK-" + str(task["extra"]["depends_on"][0])
1180             error_text = ""
1181             interfaces = task.get("depends").get(dep_id).get("extra").get("interfaces").keys()
1182             # Bear in mind that different VIM connectors might support Classifications differently.
1183             # In the case of OpenStack, only the first VNF attached to the classifier will be used
1184             # to create the Classification(s) (the "logical source port" of the "Flow Classifier").
1185             # Since the VNFFG classifier match lacks the ethertype, classification defaults to
1186             # using the IPv4 flow classifier.
1187             name = "c-%s" % task["item_id"][:8]
1188             # if not CIDR is given for the IP addresses, add /32:
1189             ip_proto = int(params.get("ip_proto"))
1190             source_ip = params.get("source_ip")
1191             destination_ip = params.get("destination_ip")
1192             source_port = params.get("source_port")
1193             destination_port = params.get("destination_port")
1194             definition = {"logical_source_port": interfaces[0]}
1195             if ip_proto:
1196                 if ip_proto == 1:
1197                     ip_proto = 'icmp'
1198                 elif ip_proto == 6:
1199                     ip_proto = 'tcp'
1200                 elif ip_proto == 17:
1201                     ip_proto = 'udp'
1202                 definition["protocol"] = ip_proto
1203             if source_ip:
1204                 if '/' not in source_ip:
1205                     source_ip += '/32'
1206                 definition["source_ip_prefix"] = source_ip
1207             if source_port:
1208                 definition["source_port_range_min"] = source_port
1209                 definition["source_port_range_max"] = source_port
1210             if destination_port:
1211                 definition["destination_port_range_min"] = destination_port
1212                 definition["destination_port_range_max"] = destination_port
1213             if destination_ip:
1214                 if '/' not in destination_ip:
1215                     destination_ip += '/32'
1216                 definition["destination_ip_prefix"] = destination_ip
1217
1218             vim_classification_id = self.vim.new_classification(
1219                 name, 'legacy_flow_classifier', definition)
1220
1221             task["extra"]["created"] = True
1222             task["extra"]["vim_status"] = "ACTIVE"
1223             task["error_msg"] = None
1224             task["status"] = "FINISHED"  # with FINISHED instead of DONE it will not be refreshing
1225             task["vim_id"] = vim_classification_id
1226             instance_element_update = {"status": "ACTIVE", "vim_classification_id": vim_classification_id,
1227                                        "error_msg": None}
1228             return instance_element_update
1229
1230         except (vimconn.vimconnException, VimThreadException) as e:
1231             self.logger.error("Error creating Classification, task=%s: %s", task_id, str(e))
1232             error_text = self._format_vim_error_msg(str(e))
1233             task["error_msg"] = error_text
1234             task["status"] = "FAILED"
1235             task["vim_id"] = None
1236             instance_element_update = {"status": "VIM_ERROR", "vim_classification_id": None, "error_msg": error_text}
1237             return instance_element_update
1238
1239     def del_classification(self, task):
1240         classification_vim_id = task["vim_id"]
1241         try:
1242             self.vim.delete_classification(classification_vim_id)
1243             task["status"] = "FINISHED"  # with FINISHED instead of DONE it will not be refreshing
1244             task["error_msg"] = None
1245             return None
1246
1247         except vimconn.vimconnException as e:
1248             task["error_msg"] = self._format_vim_error_msg(str(e))
1249             if isinstance(e, vimconn.vimconnNotFoundException):
1250                 # If not found mark as Done and fill error_msg
1251                 task["status"] = "FINISHED"  # with FINISHED instead of DONE it will not be refreshing
1252                 return None
1253             task["status"] = "FAILED"
1254             return None
1255
1256     def new_sfp(self, task):
1257         vim_sfp_id = None
1258         try:
1259             task_id = task["instance_action_id"] + "." + str(task["task_index"])
1260             depending_tasks = [task.get("depends").get("TASK-" + str(tsk_id)) for tsk_id in
1261                                task.get("extra").get("depends_on")]
1262             error_text = ""
1263             sf_id_list = []
1264             classification_id_list = []
1265             for dep in depending_tasks:
1266                 vim_id = dep.get("vim_id")
1267                 resource = dep.get("item")
1268                 if resource == "instance_sfs":
1269                     sf_id_list.append(vim_id)
1270                 elif resource == "instance_classifications":
1271                     classification_id_list.append(vim_id)
1272
1273             name = "sfp-%s" % task["item_id"][:8]
1274             # By default no form of IETF SFC Encapsulation will be used
1275             vim_sfp_id = self.vim.new_sfp(name, classification_id_list, sf_id_list, sfc_encap=False)
1276
1277             task["extra"]["created"] = True
1278             task["extra"]["vim_status"] = "ACTIVE"
1279             task["error_msg"] = None
1280             task["status"] = "FINISHED"  # with FINISHED instead of DONE it will not be refreshing
1281             task["vim_id"] = vim_sfp_id
1282             instance_element_update = {"status": "ACTIVE", "vim_sfp_id": vim_sfp_id, "error_msg": None}
1283             return instance_element_update
1284
1285         except (vimconn.vimconnException, VimThreadException) as e:
1286             self.logger.error("Error creating Service Function, task=%s: %s", task_id, str(e))
1287             error_text = self._format_vim_error_msg(str(e))
1288             task["error_msg"] = error_text
1289             task["status"] = "FAILED"
1290             task["vim_id"] = None
1291             instance_element_update = {"status": "VIM_ERROR", "vim_sfp_id": None, "error_msg": error_text}
1292             return instance_element_update
1293
1294     def del_sfp(self, task):
1295         sfp_vim_id = task["vim_id"]
1296         try:
1297             self.vim.delete_sfp(sfp_vim_id)
1298             task["status"] = "FINISHED"  # with FINISHED instead of DONE it will not be refreshing
1299             task["error_msg"] = None
1300             return None
1301
1302         except vimconn.vimconnException as e:
1303             task["error_msg"] = self._format_vim_error_msg(str(e))
1304             if isinstance(e, vimconn.vimconnNotFoundException):
1305                 # If not found mark as Done and fill error_msg
1306                 task["status"] = "FINISHED"  # with FINISHED instead of DONE it will not be refreshing
1307                 return None
1308             task["status"] = "FAILED"
1309             return None