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