NG-RO/osm_ng_ro/ns.py

   1 # -*- coding: utf-8 -*-
   2
   3 ##
   4 # Copyright 2020 Telefonica Investigacion y Desarrollo, S.A.U.
   5 # Licensed under the Apache License, Version 2.0 (the "License");
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at
   8 #
   9 #    http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an "AS IS" BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  14 # implied.
  15 # See the License for the specific language governing permissions and
  16 # limitations under the License.
  17 ##
  18
  19 from http import HTTPStatus
  20 from itertools import product
  21 import logging
  22 from random import choice as random_choice
  23 from threading import Lock
  24 from time import time
  25 from traceback import format_exc as traceback_format_exc
  26 from typing import Any, Dict, List, Optional, Tuple, Type
  27 from uuid import uuid4
  28
  29 from cryptography.hazmat.backends import default_backend as crypto_default_backend
  30 from cryptography.hazmat.primitives import serialization as crypto_serialization
  31 from cryptography.hazmat.primitives.asymmetric import rsa
  32 from jinja2 import (
  33     Environment,
  34     select_autoescape,
  35     StrictUndefined,
  36     TemplateError,
  37     TemplateNotFound,
  38     UndefinedError,
  39 )
  40 from osm_common import (
  41     dbmemory,
  42     dbmongo,
  43     fslocal,
  44     fsmongo,
  45     msgkafka,
  46     msglocal,
  47     version as common_version,
  48 )
  49 from osm_common.dbbase import DbBase, DbException
  50 from osm_common.fsbase import FsBase, FsException
  51 from osm_common.msgbase import MsgException
  52 from osm_ng_ro.ns_thread import deep_get, NsWorker, NsWorkerException
  53 from osm_ng_ro.validation import deploy_schema, validate_input
  54 import yaml
  55
  56 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
  57 min_common_version = "0.1.16"
  58
  59
  60 class NsException(Exception):
  61     def __init__(self, message, http_code=HTTPStatus.BAD_REQUEST):
  62         self.http_code = http_code
  63         super(Exception, self).__init__(message)
  64
  65
  66 def get_process_id():
  67     """
  68     Obtain a unique ID for this process. If running from inside docker, it will get docker ID. If not it
  69     will provide a random one
  70     :return: Obtained ID
  71     """
  72     # Try getting docker id. If fails, get pid
  73     try:
  74         with open("/proc/self/cgroup", "r") as f:
  75             text_id_ = f.readline()
  76             _, _, text_id = text_id_.rpartition("/")
  77             text_id = text_id.replace("\n", "")[:12]
  78
  79             if text_id:
  80                 return text_id
  81     except Exception as error:
  82         logging.exception(f"{error} occured while getting process id")
  83
  84     # Return a random id
  85     return "".join(random_choice("0123456789abcdef") for _ in range(12))
  86
  87
  88 def versiontuple(v):
  89     """utility for compare dot separate versions. Fills with zeros to proper number comparison"""
  90     filled = []
  91
  92     for point in v.split("."):
  93         filled.append(point.zfill(8))
  94
  95     return tuple(filled)
  96
  97
  98 class Ns(object):
  99     def __init__(self):
 100         self.db = None
 101         self.fs = None
 102         self.msg = None
 103         self.config = None
 104         # self.operations = None
 105         self.logger = None
 106         # ^ Getting logger inside method self.start because parent logger (ro) is not available yet.
 107         # If done now it will not be linked to parent not getting its handler and level
 108         self.map_topic = {}
 109         self.write_lock = None
 110         self.vims_assigned = {}
 111         self.next_worker = 0
 112         self.plugins = {}
 113         self.workers = []
 114         self.process_params_function_map = {
 115             "net": Ns._process_net_params,
 116             "image": Ns._process_image_params,
 117             "flavor": Ns._process_flavor_params,
 118             "vdu": Ns._process_vdu_params,
 119             "affinity-or-anti-affinity-group": Ns._process_affinity_group_params,
 120         }
 121         self.db_path_map = {
 122             "net": "vld",
 123             "image": "image",
 124             "flavor": "flavor",
 125             "vdu": "vdur",
 126             "affinity-or-anti-affinity-group": "affinity-or-anti-affinity-group",
 127         }
 128
 129     def init_db(self, target_version):
 130         pass
 131
 132     def start(self, config):
 133         """
 134         Connect to database, filesystem storage, and messaging
 135         :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
 136         :param config: Configuration of db, storage, etc
 137         :return: None
 138         """
 139         self.config = config
 140         self.config["process_id"] = get_process_id()  # used for HA identity
 141         self.logger = logging.getLogger("ro.ns")
 142
 143         # check right version of common
 144         if versiontuple(common_version) < versiontuple(min_common_version):
 145             raise NsException(
 146                 "Not compatible osm/common version '{}'. Needed '{}' or higher".format(
 147                     common_version, min_common_version
 148                 )
 149             )
 150
 151         try:
 152             if not self.db:
 153                 if config["database"]["driver"] == "mongo":
 154                     self.db = dbmongo.DbMongo()
 155                     self.db.db_connect(config["database"])
 156                 elif config["database"]["driver"] == "memory":
 157                     self.db = dbmemory.DbMemory()
 158                     self.db.db_connect(config["database"])
 159                 else:
 160                     raise NsException(
 161                         "Invalid configuration param '{}' at '[database]':'driver'".format(
 162                             config["database"]["driver"]
 163                         )
 164                     )
 165
 166             if not self.fs:
 167                 if config["storage"]["driver"] == "local":
 168                     self.fs = fslocal.FsLocal()
 169                     self.fs.fs_connect(config["storage"])
 170                 elif config["storage"]["driver"] == "mongo":
 171                     self.fs = fsmongo.FsMongo()
 172                     self.fs.fs_connect(config["storage"])
 173                 elif config["storage"]["driver"] is None:
 174                     pass
 175                 else:
 176                     raise NsException(
 177                         "Invalid configuration param '{}' at '[storage]':'driver'".format(
 178                             config["storage"]["driver"]
 179                         )
 180                     )
 181
 182             if not self.msg:
 183                 if config["message"]["driver"] == "local":
 184                     self.msg = msglocal.MsgLocal()
 185                     self.msg.connect(config["message"])
 186                 elif config["message"]["driver"] == "kafka":
 187                     self.msg = msgkafka.MsgKafka()
 188                     self.msg.connect(config["message"])
 189                 else:
 190                     raise NsException(
 191                         "Invalid configuration param '{}' at '[message]':'driver'".format(
 192                             config["message"]["driver"]
 193                         )
 194                     )
 195
 196             # TODO load workers to deal with exising database tasks
 197
 198             self.write_lock = Lock()
 199         except (DbException, FsException, MsgException) as e:
 200             raise NsException(str(e), http_code=e.http_code)
 201
 202     def get_assigned_vims(self):
 203         return list(self.vims_assigned.keys())
 204
 205     def stop(self):
 206         try:
 207             if self.db:
 208                 self.db.db_disconnect()
 209
 210             if self.fs:
 211                 self.fs.fs_disconnect()
 212
 213             if self.msg:
 214                 self.msg.disconnect()
 215
 216             self.write_lock = None
 217         except (DbException, FsException, MsgException) as e:
 218             raise NsException(str(e), http_code=e.http_code)
 219
 220         for worker in self.workers:
 221             worker.insert_task(("terminate",))
 222
 223     def _create_worker(self):
 224         """
 225         Look for a worker thread in idle status. If not found it creates one unless the number of threads reach the
 226         limit of 'server.ns_threads' configuration. If reached, it just assigns one existing thread
 227         return the index of the assigned worker thread. Worker threads are storead at self.workers
 228         """
 229         # Look for a thread in idle status
 230         worker_id = next(
 231             (
 232                 i
 233                 for i in range(len(self.workers))
 234                 if self.workers[i] and self.workers[i].idle
 235             ),
 236             None,
 237         )
 238
 239         if worker_id is not None:
 240             # unset idle status to avoid race conditions
 241             self.workers[worker_id].idle = False
 242         else:
 243             worker_id = len(self.workers)
 244
 245             if worker_id < self.config["global"]["server.ns_threads"]:
 246                 # create a new worker
 247                 self.workers.append(
 248                     NsWorker(worker_id, self.config, self.plugins, self.db)
 249                 )
 250                 self.workers[worker_id].start()
 251             else:
 252                 # reached maximum number of threads, assign VIM to an existing one
 253                 worker_id = self.next_worker
 254                 self.next_worker = (self.next_worker + 1) % self.config["global"][
 255                     "server.ns_threads"
 256                 ]
 257
 258         return worker_id
 259
 260     def assign_vim(self, target_id):
 261         with self.write_lock:
 262             return self._assign_vim(target_id)
 263
 264     def _assign_vim(self, target_id):
 265         if target_id not in self.vims_assigned:
 266             worker_id = self.vims_assigned[target_id] = self._create_worker()
 267             self.workers[worker_id].insert_task(("load_vim", target_id))
 268
 269     def reload_vim(self, target_id):
 270         # send reload_vim to the thread working with this VIM and inform all that a VIM has been changed,
 271         # this is because database VIM information is cached for threads working with SDN
 272         with self.write_lock:
 273             for worker in self.workers:
 274                 if worker and not worker.idle:
 275                     worker.insert_task(("reload_vim", target_id))
 276
 277     def unload_vim(self, target_id):
 278         with self.write_lock:
 279             return self._unload_vim(target_id)
 280
 281     def _unload_vim(self, target_id):
 282         if target_id in self.vims_assigned:
 283             worker_id = self.vims_assigned[target_id]
 284             self.workers[worker_id].insert_task(("unload_vim", target_id))
 285             del self.vims_assigned[target_id]
 286
 287     def check_vim(self, target_id):
 288         with self.write_lock:
 289             if target_id in self.vims_assigned:
 290                 worker_id = self.vims_assigned[target_id]
 291             else:
 292                 worker_id = self._create_worker()
 293
 294         worker = self.workers[worker_id]
 295         worker.insert_task(("check_vim", target_id))
 296
 297     def unload_unused_vims(self):
 298         with self.write_lock:
 299             vims_to_unload = []
 300
 301             for target_id in self.vims_assigned:
 302                 if not self.db.get_one(
 303                     "ro_tasks",
 304                     q_filter={
 305                         "target_id": target_id,
 306                         "tasks.status": ["SCHEDULED", "BUILD", "DONE", "FAILED"],
 307                     },
 308                     fail_on_empty=False,
 309                 ):
 310                     vims_to_unload.append(target_id)
 311
 312             for target_id in vims_to_unload:
 313                 self._unload_vim(target_id)
 314
 315     @staticmethod
 316     def _get_cloud_init(
 317         db: Type[DbBase],
 318         fs: Type[FsBase],
 319         location: str,
 320     ) -> str:
 321         """This method reads cloud init from a file.
 322
 323         Note: Not used as cloud init content is provided in the http body.
 324
 325         Args:
 326             db (Type[DbBase]): [description]
 327             fs (Type[FsBase]): [description]
 328             location (str): can be 'vnfr_id:file:file_name' or 'vnfr_id:vdu:vdu_idex'
 329
 330         Raises:
 331             NsException: [description]
 332             NsException: [description]
 333
 334         Returns:
 335             str: [description]
 336         """
 337         vnfd_id, _, other = location.partition(":")
 338         _type, _, name = other.partition(":")
 339         vnfd = db.get_one("vnfds", {"_id": vnfd_id})
 340
 341         if _type == "file":
 342             base_folder = vnfd["_admin"]["storage"]
 343             cloud_init_file = "{}/{}/cloud_init/{}".format(
 344                 base_folder["folder"], base_folder["pkg-dir"], name
 345             )
 346
 347             if not fs:
 348                 raise NsException(
 349                     "Cannot read file '{}'. Filesystem not loaded, change configuration at storage.driver".format(
 350                         cloud_init_file
 351                     )
 352                 )
 353
 354             with fs.file_open(cloud_init_file, "r") as ci_file:
 355                 cloud_init_content = ci_file.read()
 356         elif _type == "vdu":
 357             cloud_init_content = vnfd["vdu"][int(name)]["cloud-init"]
 358         else:
 359             raise NsException("Mismatch descriptor for cloud init: {}".format(location))
 360
 361         return cloud_init_content
 362
 363     @staticmethod
 364     def _parse_jinja2(
 365         cloud_init_content: str,
 366         params: Dict[str, Any],
 367         context: str,
 368     ) -> str:
 369         """Function that processes the cloud init to replace Jinja2 encoded parameters.
 370
 371         Args:
 372             cloud_init_content (str): [description]
 373             params (Dict[str, Any]): [description]
 374             context (str): [description]
 375
 376         Raises:
 377             NsException: [description]
 378             NsException: [description]
 379
 380         Returns:
 381             str: [description]
 382         """
 383         try:
 384             env = Environment(
 385                 undefined=StrictUndefined,
 386                 autoescape=select_autoescape(default_for_string=True, default=True),
 387             )
 388             template = env.from_string(cloud_init_content)
 389
 390             return template.render(params or {})
 391         except UndefinedError as e:
 392             raise NsException(
 393                 "Variable '{}' defined at vnfd='{}' must be provided in the instantiation parameters"
 394                 "inside the 'additionalParamsForVnf' block".format(e, context)
 395             )
 396         except (TemplateError, TemplateNotFound) as e:
 397             raise NsException(
 398                 "Error parsing Jinja2 to cloud-init content at vnfd='{}': {}".format(
 399                     context, e
 400                 )
 401             )
 402
 403     def _create_db_ro_nsrs(self, nsr_id, now):
 404         try:
 405             key = rsa.generate_private_key(
 406                 backend=crypto_default_backend(), public_exponent=65537, key_size=2048
 407             )
 408             private_key = key.private_bytes(
 409                 crypto_serialization.Encoding.PEM,
 410                 crypto_serialization.PrivateFormat.PKCS8,
 411                 crypto_serialization.NoEncryption(),
 412             )
 413             public_key = key.public_key().public_bytes(
 414                 crypto_serialization.Encoding.OpenSSH,
 415                 crypto_serialization.PublicFormat.OpenSSH,
 416             )
 417             private_key = private_key.decode("utf8")
 418             # Change first line because Paramiko needs a explicit start with 'BEGIN RSA PRIVATE KEY'
 419             i = private_key.find("\n")
 420             private_key = "-----BEGIN RSA PRIVATE KEY-----" + private_key[i:]
 421             public_key = public_key.decode("utf8")
 422         except Exception as e:
 423             raise NsException("Cannot create ssh-keys: {}".format(e))
 424
 425         schema_version = "1.1"
 426         private_key_encrypted = self.db.encrypt(
 427             private_key, schema_version=schema_version, salt=nsr_id
 428         )
 429         db_content = {
 430             "_id": nsr_id,
 431             "_admin": {
 432                 "created": now,
 433                 "modified": now,
 434                 "schema_version": schema_version,
 435             },
 436             "public_key": public_key,
 437             "private_key": private_key_encrypted,
 438             "actions": [],
 439         }
 440         self.db.create("ro_nsrs", db_content)
 441
 442         return db_content
 443
 444     @staticmethod
 445     def _create_task(
 446         deployment_info: Dict[str, Any],
 447         target_id: str,
 448         item: str,
 449         action: str,
 450         target_record: str,
 451         target_record_id: str,
 452         extra_dict: Dict[str, Any] = None,
 453     ) -> Dict[str, Any]:
 454         """Function to create task dict from deployment information.
 455
 456         Args:
 457             deployment_info (Dict[str, Any]): [description]
 458             target_id (str): [description]
 459             item (str): [description]
 460             action (str): [description]
 461             target_record (str): [description]
 462             target_record_id (str): [description]
 463             extra_dict (Dict[str, Any], optional): [description]. Defaults to None.
 464
 465         Returns:
 466             Dict[str, Any]: [description]
 467         """
 468         task = {
 469             "target_id": target_id,  # it will be removed before pushing at database
 470             "action_id": deployment_info.get("action_id"),
 471             "nsr_id": deployment_info.get("nsr_id"),
 472             "task_id": f"{deployment_info.get('action_id')}:{deployment_info.get('task_index')}",
 473             "status": "SCHEDULED",
 474             "action": action,
 475             "item": item,
 476             "target_record": target_record,
 477             "target_record_id": target_record_id,
 478         }
 479
 480         if extra_dict:
 481             task.update(extra_dict)  # params, find_params, depends_on
 482
 483         deployment_info["task_index"] = deployment_info.get("task_index", 0) + 1
 484
 485         return task
 486
 487     @staticmethod
 488     def _create_ro_task(
 489         target_id: str,
 490         task: Dict[str, Any],
 491     ) -> Dict[str, Any]:
 492         """Function to create an RO task from task information.
 493
 494         Args:
 495             target_id (str): [description]
 496             task (Dict[str, Any]): [description]
 497
 498         Returns:
 499             Dict[str, Any]: [description]
 500         """
 501         now = time()
 502
 503         _id = task.get("task_id")
 504         db_ro_task = {
 505             "_id": _id,
 506             "locked_by": None,
 507             "locked_at": 0.0,
 508             "target_id": target_id,
 509             "vim_info": {
 510                 "created": False,
 511                 "created_items": None,
 512                 "vim_id": None,
 513                 "vim_name": None,
 514                 "vim_status": None,
 515                 "vim_details": None,
 516                 "vim_message": None,
 517                 "refresh_at": None,
 518             },
 519             "modified_at": now,
 520             "created_at": now,
 521             "to_check_at": now,
 522             "tasks": [task],
 523         }
 524
 525         return db_ro_task
 526
 527     @staticmethod
 528     def _process_image_params(
 529         target_image: Dict[str, Any],
 530         indata: Dict[str, Any],
 531         vim_info: Dict[str, Any],
 532         target_record_id: str,
 533         **kwargs: Dict[str, Any],
 534     ) -> Dict[str, Any]:
 535         """Function to process VDU image parameters.
 536
 537         Args:
 538             target_image (Dict[str, Any]): [description]
 539             indata (Dict[str, Any]): [description]
 540             vim_info (Dict[str, Any]): [description]
 541             target_record_id (str): [description]
 542
 543         Returns:
 544             Dict[str, Any]: [description]
 545         """
 546         find_params = {}
 547
 548         if target_image.get("image"):
 549             find_params["filter_dict"] = {"name": target_image.get("image")}
 550
 551         if target_image.get("vim_image_id"):
 552             find_params["filter_dict"] = {"id": target_image.get("vim_image_id")}
 553
 554         if target_image.get("image_checksum"):
 555             find_params["filter_dict"] = {
 556                 "checksum": target_image.get("image_checksum")
 557             }
 558
 559         return {"find_params": find_params}
 560
 561     @staticmethod
 562     def _get_resource_allocation_params(
 563         quota_descriptor: Dict[str, Any],
 564     ) -> Dict[str, Any]:
 565         """Read the quota_descriptor from vnfd and fetch the resource allocation properties from the
 566         descriptor object.
 567
 568         Args:
 569             quota_descriptor (Dict[str, Any]): cpu/mem/vif/disk-io quota descriptor
 570
 571         Returns:
 572             Dict[str, Any]: quota params for limit, reserve, shares from the descriptor object
 573         """
 574         quota = {}
 575
 576         if quota_descriptor.get("limit"):
 577             quota["limit"] = int(quota_descriptor["limit"])
 578
 579         if quota_descriptor.get("reserve"):
 580             quota["reserve"] = int(quota_descriptor["reserve"])
 581
 582         if quota_descriptor.get("shares"):
 583             quota["shares"] = int(quota_descriptor["shares"])
 584
 585         return quota
 586
 587     @staticmethod
 588     def _process_guest_epa_quota_params(
 589         guest_epa_quota: Dict[str, Any],
 590         epa_vcpu_set: bool,
 591     ) -> Dict[str, Any]:
 592         """Function to extract the guest epa quota parameters.
 593
 594         Args:
 595             guest_epa_quota (Dict[str, Any]): [description]
 596             epa_vcpu_set (bool): [description]
 597
 598         Returns:
 599             Dict[str, Any]: [description]
 600         """
 601         result = {}
 602
 603         if guest_epa_quota.get("cpu-quota") and not epa_vcpu_set:
 604             cpuquota = Ns._get_resource_allocation_params(
 605                 guest_epa_quota.get("cpu-quota")
 606             )
 607
 608             if cpuquota:
 609                 result["cpu-quota"] = cpuquota
 610
 611         if guest_epa_quota.get("mem-quota"):
 612             vduquota = Ns._get_resource_allocation_params(
 613                 guest_epa_quota.get("mem-quota")
 614             )
 615
 616             if vduquota:
 617                 result["mem-quota"] = vduquota
 618
 619         if guest_epa_quota.get("disk-io-quota"):
 620             diskioquota = Ns._get_resource_allocation_params(
 621                 guest_epa_quota.get("disk-io-quota")
 622             )
 623
 624             if diskioquota:
 625                 result["disk-io-quota"] = diskioquota
 626
 627         if guest_epa_quota.get("vif-quota"):
 628             vifquota = Ns._get_resource_allocation_params(
 629                 guest_epa_quota.get("vif-quota")
 630             )
 631
 632             if vifquota:
 633                 result["vif-quota"] = vifquota
 634
 635         return result
 636
 637     @staticmethod
 638     def _process_guest_epa_numa_params(
 639         guest_epa_quota: Dict[str, Any],
 640     ) -> Tuple[Dict[str, Any], bool]:
 641         """[summary]
 642
 643         Args:
 644             guest_epa_quota (Dict[str, Any]): [description]
 645
 646         Returns:
 647             Tuple[Dict[str, Any], bool]: [description]
 648         """
 649         numa = {}
 650         numa_list = []
 651         epa_vcpu_set = False
 652
 653         if guest_epa_quota.get("numa-node-policy"):
 654             numa_node_policy = guest_epa_quota.get("numa-node-policy")
 655
 656             if numa_node_policy.get("node"):
 657                 for numa_node in numa_node_policy["node"]:
 658                     vcpu_list = []
 659                     if numa_node.get("id"):
 660                         numa["id"] = int(numa_node["id"])
 661
 662                     if numa_node.get("vcpu"):
 663                         for vcpu in numa_node.get("vcpu"):
 664                             vcpu_id = int(vcpu.get("id"))
 665                             vcpu_list.append(vcpu_id)
 666                         numa["vcpu"] = vcpu_list
 667
 668                     if numa_node.get("num-cores"):
 669                         numa["cores"] = numa_node["num-cores"]
 670                         epa_vcpu_set = True
 671
 672                     paired_threads = numa_node.get("paired-threads", {})
 673                     if paired_threads.get("num-paired-threads"):
 674                         numa["paired_threads"] = int(
 675                             numa_node["paired-threads"]["num-paired-threads"]
 676                         )
 677                         epa_vcpu_set = True
 678
 679                     if paired_threads.get("paired-thread-ids"):
 680                         numa["paired-threads-id"] = []
 681
 682                         for pair in paired_threads["paired-thread-ids"]:
 683                             numa["paired-threads-id"].append(
 684                                 (
 685                                     str(pair["thread-a"]),
 686                                     str(pair["thread-b"]),
 687                                 )
 688                             )
 689
 690                     if numa_node.get("num-threads"):
 691                         numa["threads"] = int(numa_node["num-threads"])
 692                         epa_vcpu_set = True
 693
 694                     if numa_node.get("memory-mb"):
 695                         numa["memory"] = max(int(int(numa_node["memory-mb"]) / 1024), 1)
 696
 697                     numa_list.append(numa)
 698                     numa = {}
 699
 700         return numa_list, epa_vcpu_set
 701
 702     @staticmethod
 703     def _process_guest_epa_cpu_pinning_params(
 704         guest_epa_quota: Dict[str, Any],
 705         vcpu_count: int,
 706         epa_vcpu_set: bool,
 707     ) -> Tuple[Dict[str, Any], bool]:
 708         """[summary]
 709
 710         Args:
 711             guest_epa_quota (Dict[str, Any]): [description]
 712             vcpu_count (int): [description]
 713             epa_vcpu_set (bool): [description]
 714
 715         Returns:
 716             Tuple[Dict[str, Any], bool]: [description]
 717         """
 718         numa = {}
 719         local_epa_vcpu_set = epa_vcpu_set
 720
 721         if (
 722             guest_epa_quota.get("cpu-pinning-policy") == "DEDICATED"
 723             and not epa_vcpu_set
 724         ):
 725             # Pinning policy "REQUIRE" uses threads as host should support SMT architecture
 726             # Pinning policy "ISOLATE" uses cores as host should not support SMT architecture
 727             # Pinning policy "PREFER" uses threads in case host supports SMT architecture
 728             numa[
 729                 "cores"
 730                 if guest_epa_quota.get("cpu-thread-pinning-policy") == "ISOLATE"
 731                 else "threads"
 732             ] = max(vcpu_count, 1)
 733             local_epa_vcpu_set = True
 734
 735         return numa, local_epa_vcpu_set
 736
 737     @staticmethod
 738     def _process_epa_params(
 739         target_flavor: Dict[str, Any],
 740     ) -> Dict[str, Any]:
 741         """[summary]
 742
 743         Args:
 744             target_flavor (Dict[str, Any]): [description]
 745
 746         Returns:
 747             Dict[str, Any]: [description]
 748         """
 749         extended = {}
 750         numa = {}
 751         numa_list = []
 752
 753         if target_flavor.get("guest-epa"):
 754             guest_epa = target_flavor["guest-epa"]
 755
 756             numa_list, epa_vcpu_set = Ns._process_guest_epa_numa_params(
 757                 guest_epa_quota=guest_epa
 758             )
 759
 760             if guest_epa.get("mempage-size"):
 761                 extended["mempage-size"] = guest_epa.get("mempage-size")
 762
 763             if guest_epa.get("cpu-pinning-policy"):
 764                 extended["cpu-pinning-policy"] = guest_epa.get("cpu-pinning-policy")
 765
 766             if guest_epa.get("cpu-thread-pinning-policy"):
 767                 extended["cpu-thread-pinning-policy"] = guest_epa.get(
 768                     "cpu-thread-pinning-policy"
 769                 )
 770
 771             if guest_epa.get("numa-node-policy"):
 772                 if guest_epa.get("numa-node-policy").get("mem-policy"):
 773                     extended["mem-policy"] = guest_epa.get("numa-node-policy").get(
 774                         "mem-policy"
 775                     )
 776
 777             tmp_numa, epa_vcpu_set = Ns._process_guest_epa_cpu_pinning_params(
 778                 guest_epa_quota=guest_epa,
 779                 vcpu_count=int(target_flavor.get("vcpu-count", 1)),
 780                 epa_vcpu_set=epa_vcpu_set,
 781             )
 782             for numa in numa_list:
 783                 numa.update(tmp_numa)
 784
 785             extended.update(
 786                 Ns._process_guest_epa_quota_params(
 787                     guest_epa_quota=guest_epa,
 788                     epa_vcpu_set=epa_vcpu_set,
 789                 )
 790             )
 791
 792         if numa:
 793             extended["numas"] = numa_list
 794
 795         return extended
 796
 797     @staticmethod
 798     def _process_flavor_params(
 799         target_flavor: Dict[str, Any],
 800         indata: Dict[str, Any],
 801         vim_info: Dict[str, Any],
 802         target_record_id: str,
 803         **kwargs: Dict[str, Any],
 804     ) -> Dict[str, Any]:
 805         """[summary]
 806
 807         Args:
 808             target_flavor (Dict[str, Any]): [description]
 809             indata (Dict[str, Any]): [description]
 810             vim_info (Dict[str, Any]): [description]
 811             target_record_id (str): [description]
 812
 813         Returns:
 814             Dict[str, Any]: [description]
 815         """
 816         db = kwargs.get("db")
 817         target_vdur = {}
 818
 819         flavor_data = {
 820             "disk": int(target_flavor["storage-gb"]),
 821             "ram": int(target_flavor["memory-mb"]),
 822             "vcpus": int(target_flavor["vcpu-count"]),
 823         }
 824
 825         for vnf in indata.get("vnf", []):
 826             for vdur in vnf.get("vdur", []):
 827                 if vdur.get("ns-flavor-id") == target_flavor.get("id"):
 828                     target_vdur = vdur
 829
 830         if db and isinstance(indata.get("vnf"), list):
 831             vnfd_id = indata.get("vnf")[0].get("vnfd-id")
 832             vnfd = db.get_one("vnfds", {"_id": vnfd_id})
 833             # check if there is persistent root disk
 834             for vdu in vnfd.get("vdu", ()):
 835                 if vdu["name"] == target_vdur.get("vdu-name"):
 836                     for vsd in vnfd.get("virtual-storage-desc", ()):
 837                         if vsd.get("id") == vdu.get("virtual-storage-desc", [[]])[0]:
 838                             root_disk = vsd
 839                             if (
 840                                 root_disk.get("type-of-storage")
 841                                 == "persistent-storage:persistent-storage"
 842                             ):
 843                                 flavor_data["disk"] = 0
 844
 845         for storage in target_vdur.get("virtual-storages", []):
 846             if (
 847                 storage.get("type-of-storage")
 848                 == "etsi-nfv-descriptors:ephemeral-storage"
 849             ):
 850                 flavor_data["ephemeral"] = int(storage.get("size-of-storage", 0))
 851             elif storage.get("type-of-storage") == "etsi-nfv-descriptors:swap-storage":
 852                 flavor_data["swap"] = int(storage.get("size-of-storage", 0))
 853
 854         extended = Ns._process_epa_params(target_flavor)
 855         if extended:
 856             flavor_data["extended"] = extended
 857
 858         extra_dict = {"find_params": {"flavor_data": flavor_data}}
 859         flavor_data_name = flavor_data.copy()
 860         flavor_data_name["name"] = target_flavor["name"]
 861         extra_dict["params"] = {"flavor_data": flavor_data_name}
 862
 863         return extra_dict
 864
 865     @staticmethod
 866     def _ip_profile_to_ro(
 867         ip_profile: Dict[str, Any],
 868     ) -> Dict[str, Any]:
 869         """[summary]
 870
 871         Args:
 872             ip_profile (Dict[str, Any]): [description]
 873
 874         Returns:
 875             Dict[str, Any]: [description]
 876         """
 877         if not ip_profile:
 878             return None
 879
 880         ro_ip_profile = {
 881             "ip_version": "IPv4"
 882             if "v4" in ip_profile.get("ip-version", "ipv4")
 883             else "IPv6",
 884             "subnet_address": ip_profile.get("subnet-address"),
 885             "gateway_address": ip_profile.get("gateway-address"),
 886             "dhcp_enabled": ip_profile.get("dhcp-params", {}).get("enabled", False),
 887             "dhcp_start_address": ip_profile.get("dhcp-params", {}).get(
 888                 "start-address", None
 889             ),
 890             "dhcp_count": ip_profile.get("dhcp-params", {}).get("count", None),
 891         }
 892
 893         if ip_profile.get("dns-server"):
 894             ro_ip_profile["dns_address"] = ";".join(
 895                 [v["address"] for v in ip_profile["dns-server"] if v.get("address")]
 896             )
 897
 898         if ip_profile.get("security-group"):
 899             ro_ip_profile["security_group"] = ip_profile["security-group"]
 900
 901         return ro_ip_profile
 902
 903     @staticmethod
 904     def _process_net_params(
 905         target_vld: Dict[str, Any],
 906         indata: Dict[str, Any],
 907         vim_info: Dict[str, Any],
 908         target_record_id: str,
 909         **kwargs: Dict[str, Any],
 910     ) -> Dict[str, Any]:
 911         """Function to process network parameters.
 912
 913         Args:
 914             target_vld (Dict[str, Any]): [description]
 915             indata (Dict[str, Any]): [description]
 916             vim_info (Dict[str, Any]): [description]
 917             target_record_id (str): [description]
 918
 919         Returns:
 920             Dict[str, Any]: [description]
 921         """
 922         extra_dict = {}
 923
 924         if vim_info.get("sdn"):
 925             # vnf_preffix = "vnfrs:{}".format(vnfr_id)
 926             # ns_preffix = "nsrs:{}".format(nsr_id)
 927             # remove the ending ".sdn
 928             vld_target_record_id, _, _ = target_record_id.rpartition(".")
 929             extra_dict["params"] = {
 930                 k: vim_info[k]
 931                 for k in ("sdn-ports", "target_vim", "vlds", "type")
 932                 if vim_info.get(k)
 933             }
 934
 935             # TODO needed to add target_id in the dependency.
 936             if vim_info.get("target_vim"):
 937                 extra_dict["depends_on"] = [
 938                     f"{vim_info.get('target_vim')} {vld_target_record_id}"
 939                 ]
 940
 941             return extra_dict
 942
 943         if vim_info.get("vim_network_name"):
 944             extra_dict["find_params"] = {
 945                 "filter_dict": {
 946                     "name": vim_info.get("vim_network_name"),
 947                 },
 948             }
 949         elif vim_info.get("vim_network_id"):
 950             extra_dict["find_params"] = {
 951                 "filter_dict": {
 952                     "id": vim_info.get("vim_network_id"),
 953                 },
 954             }
 955         elif target_vld.get("mgmt-network") and not vim_info.get("provider_network"):
 956             extra_dict["find_params"] = {
 957                 "mgmt": True,
 958                 "name": target_vld["id"],
 959             }
 960         else:
 961             # create
 962             extra_dict["params"] = {
 963                 "net_name": (
 964                     f"{indata.get('name')[:16]}-{target_vld.get('name', target_vld.get('id'))[:16]}"
 965                 ),
 966                 "ip_profile": Ns._ip_profile_to_ro(vim_info.get("ip_profile")),
 967                 "provider_network_profile": vim_info.get("provider_network"),
 968             }
 969
 970             if not target_vld.get("underlay"):
 971                 extra_dict["params"]["net_type"] = "bridge"
 972             else:
 973                 extra_dict["params"]["net_type"] = (
 974                     "ptp" if target_vld.get("type") == "ELINE" else "data"
 975                 )
 976
 977         return extra_dict
 978
 979     @staticmethod
 980     def find_persistent_root_volumes(
 981         vnfd: dict,
 982         target_vdu: dict,
 983         vdu_instantiation_volumes_list: list,
 984         disk_list: list,
 985     ) -> Dict[str, any]:
 986         """Find the persistent root volumes and add them to the disk_list
 987         by parsing the instantiation parameters.
 988
 989         Args:
 990             vnfd    (dict):                                 VNF descriptor
 991             target_vdu      (dict):                         processed VDU
 992             vdu_instantiation_volumes_list  (list):         instantiation parameters for the each VDU as a list
 993             disk_list   (list):                             to be filled up
 994
 995         Returns:
 996             persistent_root_disk    (dict):                 Details of persistent root disk
 997
 998         """
 999         persistent_root_disk = {}
1000         # There can be only one root disk, when we find it, it will return the result
1001
1002         for vdu, vsd in product(
1003             vnfd.get("vdu", ()), vnfd.get("virtual-storage-desc", ())
1004         ):
1005             if (
1006                 vdu["name"] == target_vdu["vdu-name"]
1007                 and vsd.get("id") == vdu.get("virtual-storage-desc", [[]])[0]
1008             ):
1009                 root_disk = vsd
1010                 if (
1011                     root_disk.get("type-of-storage")
1012                     == "persistent-storage:persistent-storage"
1013                 ):
1014                     for vdu_volume in vdu_instantiation_volumes_list:
1015                         if (
1016                             vdu_volume["vim-volume-id"]
1017                             and root_disk["id"] == vdu_volume["name"]
1018                         ):
1019                             persistent_root_disk[vsd["id"]] = {
1020                                 "vim_volume_id": vdu_volume["vim-volume-id"],
1021                                 "image_id": vdu.get("sw-image-desc"),
1022                             }
1023
1024                             disk_list.append(persistent_root_disk[vsd["id"]])
1025
1026                             return persistent_root_disk
1027
1028                     else:
1029                         if root_disk.get("size-of-storage"):
1030                             persistent_root_disk[vsd["id"]] = {
1031                                 "image_id": vdu.get("sw-image-desc"),
1032                                 "size": root_disk.get("size-of-storage"),
1033                             }
1034
1035                             disk_list.append(persistent_root_disk[vsd["id"]])
1036
1037                             return persistent_root_disk
1038
1039     @staticmethod
1040     def find_persistent_volumes(
1041         persistent_root_disk: dict,
1042         target_vdu: dict,
1043         vdu_instantiation_volumes_list: list,
1044         disk_list: list,
1045     ) -> None:
1046         """Find the ordinary persistent volumes and add them to the disk_list
1047         by parsing the instantiation parameters.
1048
1049         Args:
1050             persistent_root_disk:   persistent root disk dictionary
1051             target_vdu: processed VDU
1052             vdu_instantiation_volumes_list: instantiation parameters for the each VDU as a list
1053             disk_list:  to be filled up
1054
1055         """
1056         # Find the ordinary volumes which are not added to the persistent_root_disk
1057         persistent_disk = {}
1058         for disk in target_vdu.get("virtual-storages", {}):
1059             if (
1060                 disk.get("type-of-storage") == "persistent-storage:persistent-storage"
1061                 and disk["id"] not in persistent_root_disk.keys()
1062             ):
1063                 for vdu_volume in vdu_instantiation_volumes_list:
1064                     if vdu_volume["vim-volume-id"] and disk["id"] == vdu_volume["name"]:
1065                         persistent_disk[disk["id"]] = {
1066                             "vim_volume_id": vdu_volume["vim-volume-id"],
1067                         }
1068                         disk_list.append(persistent_disk[disk["id"]])
1069
1070                 else:
1071                     if disk["id"] not in persistent_disk.keys():
1072                         persistent_disk[disk["id"]] = {
1073                             "size": disk.get("size-of-storage"),
1074                         }
1075                         disk_list.append(persistent_disk[disk["id"]])
1076
1077     @staticmethod
1078     def _sort_vdu_interfaces(target_vdu: dict) -> None:
1079         """Sort the interfaces according to position number.
1080
1081         Args:
1082             target_vdu  (dict):     Details of VDU to be created
1083
1084         """
1085         # If the position info is provided for all the interfaces, it will be sorted
1086         # according to position number ascendingly.
1087         sorted_interfaces = sorted(
1088             target_vdu["interfaces"],
1089             key=lambda x: (x.get("position") is None, x.get("position")),
1090         )
1091         target_vdu["interfaces"] = sorted_interfaces
1092
1093     @staticmethod
1094     def _partially_locate_vdu_interfaces(target_vdu: dict) -> None:
1095         """Only place the interfaces which has specific position.
1096
1097         Args:
1098             target_vdu  (dict):     Details of VDU to be created
1099
1100         """
1101         # If the position info is provided for some interfaces but not all of them, the interfaces
1102         # which has specific position numbers will be placed and others' positions will not be taken care.
1103         if any(
1104             i.get("position") + 1
1105             for i in target_vdu["interfaces"]
1106             if i.get("position") is not None
1107         ):
1108             n = len(target_vdu["interfaces"])
1109             sorted_interfaces = [-1] * n
1110             k, m = 0, 0
1111
1112             while k < n:
1113                 if target_vdu["interfaces"][k].get("position") is not None:
1114                     if any(i.get("position") == 0 for i in target_vdu["interfaces"]):
1115                         idx = target_vdu["interfaces"][k]["position"] + 1
1116                     else:
1117                         idx = target_vdu["interfaces"][k]["position"]
1118                     sorted_interfaces[idx - 1] = target_vdu["interfaces"][k]
1119                 k += 1
1120
1121             while m < n:
1122                 if target_vdu["interfaces"][m].get("position") is None:
1123                     idy = sorted_interfaces.index(-1)
1124                     sorted_interfaces[idy] = target_vdu["interfaces"][m]
1125                 m += 1
1126
1127             target_vdu["interfaces"] = sorted_interfaces
1128
1129     @staticmethod
1130     def _prepare_vdu_cloud_init(
1131         target_vdu: dict, vdu2cloud_init: dict, db: object, fs: object
1132     ) -> Dict:
1133         """Fill cloud_config dict with cloud init details.
1134
1135         Args:
1136             target_vdu  (dict):         Details of VDU to be created
1137             vdu2cloud_init  (dict):     Cloud init dict
1138             db  (object):               DB object
1139             fs  (object):               FS object
1140
1141         Returns:
1142             cloud_config (dict):        Cloud config details of VDU
1143
1144         """
1145         # cloud config
1146         cloud_config = {}
1147
1148         if target_vdu.get("cloud-init"):
1149             if target_vdu["cloud-init"] not in vdu2cloud_init:
1150                 vdu2cloud_init[target_vdu["cloud-init"]] = Ns._get_cloud_init(
1151                     db=db,
1152                     fs=fs,
1153                     location=target_vdu["cloud-init"],
1154                 )
1155
1156             cloud_content_ = vdu2cloud_init[target_vdu["cloud-init"]]
1157             cloud_config["user-data"] = Ns._parse_jinja2(
1158                 cloud_init_content=cloud_content_,
1159                 params=target_vdu.get("additionalParams"),
1160                 context=target_vdu["cloud-init"],
1161             )
1162
1163         if target_vdu.get("boot-data-drive"):
1164             cloud_config["boot-data-drive"] = target_vdu.get("boot-data-drive")
1165
1166         return cloud_config
1167
1168     @staticmethod
1169     def _check_vld_information_of_interfaces(
1170         interface: dict, ns_preffix: str, vnf_preffix: str
1171     ) -> Optional[str]:
1172         """Prepare the net_text by the virtual link information for vnf and ns level.
1173         Args:
1174             interface   (dict):         Interface details
1175             ns_preffix  (str):          Prefix of NS
1176             vnf_preffix (str):          Prefix of VNF
1177
1178         Returns:
1179             net_text    (str):          information of net
1180
1181         """
1182         net_text = ""
1183         if interface.get("ns-vld-id"):
1184             net_text = ns_preffix + ":vld." + interface["ns-vld-id"]
1185         elif interface.get("vnf-vld-id"):
1186             net_text = vnf_preffix + ":vld." + interface["vnf-vld-id"]
1187
1188         return net_text
1189
1190     @staticmethod
1191     def _prepare_interface_port_security(interface: dict) -> None:
1192         """
1193
1194         Args:
1195             interface   (dict):     Interface details
1196
1197         """
1198         if "port-security-enabled" in interface:
1199             interface["port_security"] = interface.pop("port-security-enabled")
1200
1201         if "port-security-disable-strategy" in interface:
1202             interface["port_security_disable_strategy"] = interface.pop(
1203                 "port-security-disable-strategy"
1204             )
1205
1206     @staticmethod
1207     def _create_net_item_of_interface(interface: dict, net_text: str) -> dict:
1208         """Prepare net item including name, port security, floating ip etc.
1209
1210         Args:
1211             interface   (dict):         Interface details
1212             net_text    (str):          information of net
1213
1214         Returns:
1215             net_item    (dict):         Dict including net details
1216
1217         """
1218
1219         net_item = {
1220             x: v
1221             for x, v in interface.items()
1222             if x
1223             in (
1224                 "name",
1225                 "vpci",
1226                 "port_security",
1227                 "port_security_disable_strategy",
1228                 "floating_ip",
1229             )
1230         }
1231         net_item["net_id"] = "TASK-" + net_text
1232         net_item["type"] = "virtual"
1233
1234         return net_item
1235
1236     @staticmethod
1237     def _prepare_type_of_interface(
1238         interface: dict, tasks_by_target_record_id: dict, net_text: str, net_item: dict
1239     ) -> None:
1240         """Fill the net item type by interface type such as SR-IOV, OM-MGMT, bridge etc.
1241
1242         Args:
1243             interface   (dict):                     Interface details
1244             tasks_by_target_record_id   (dict):     Task details
1245             net_text    (str):                      information of net
1246             net_item    (dict):                     Dict including net details
1247
1248         """
1249         # TODO mac_address: used for  SR-IOV ifaces #TODO for other types
1250         # TODO floating_ip: True/False (or it can be None)
1251
1252         if interface.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
1253             # Mark the net create task as type data
1254             if deep_get(
1255                 tasks_by_target_record_id,
1256                 net_text,
1257                 "extra_dict",
1258                 "params",
1259                 "net_type",
1260             ):
1261                 tasks_by_target_record_id[net_text]["extra_dict"]["params"][
1262                     "net_type"
1263                 ] = "data"
1264
1265             net_item["use"] = "data"
1266             net_item["model"] = interface["type"]
1267             net_item["type"] = interface["type"]
1268
1269         elif (
1270             interface.get("type") == "OM-MGMT"
1271             or interface.get("mgmt-interface")
1272             or interface.get("mgmt-vnf")
1273         ):
1274             net_item["use"] = "mgmt"
1275
1276         else:
1277             # If interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"):
1278             net_item["use"] = "bridge"
1279             net_item["model"] = interface.get("type")
1280
1281     @staticmethod
1282     def _prepare_vdu_interfaces(
1283         target_vdu: dict,
1284         extra_dict: dict,
1285         ns_preffix: str,
1286         vnf_preffix: str,
1287         logger: object,
1288         tasks_by_target_record_id: dict,
1289         net_list: list,
1290     ) -> None:
1291         """Prepare the net_item and add net_list, add mgmt interface to extra_dict.
1292
1293         Args:
1294             target_vdu  (dict):                             VDU to be created
1295             extra_dict  (dict):                             Dictionary to be filled
1296             ns_preffix  (str):                              NS prefix as string
1297             vnf_preffix (str):                              VNF prefix as string
1298             logger  (object):                               Logger Object
1299             tasks_by_target_record_id  (dict):              Task details
1300             net_list    (list):                             Net list of VDU
1301         """
1302         for iface_index, interface in enumerate(target_vdu["interfaces"]):
1303             net_text = Ns._check_vld_information_of_interfaces(
1304                 interface, ns_preffix, vnf_preffix
1305             )
1306             if not net_text:
1307                 # Interface not connected to any vld
1308                 logger.error(
1309                     "Interface {} from vdu {} not connected to any vld".format(
1310                         iface_index, target_vdu["vdu-name"]
1311                     )
1312                 )
1313                 continue
1314
1315             extra_dict["depends_on"].append(net_text)
1316
1317             Ns._prepare_interface_port_security(interface)
1318
1319             net_item = Ns._create_net_item_of_interface(interface, net_text)
1320
1321             Ns._prepare_type_of_interface(
1322                 interface, tasks_by_target_record_id, net_text, net_item
1323             )
1324
1325             if interface.get("ip-address"):
1326                 net_item["ip_address"] = interface["ip-address"]
1327
1328             if interface.get("mac-address"):
1329                 net_item["mac_address"] = interface["mac-address"]
1330
1331             net_list.append(net_item)
1332
1333             if interface.get("mgmt-vnf"):
1334                 extra_dict["mgmt_vnf_interface"] = iface_index
1335             elif interface.get("mgmt-interface"):
1336                 extra_dict["mgmt_vdu_interface"] = iface_index
1337
1338     @staticmethod
1339     def _prepare_vdu_ssh_keys(
1340         target_vdu: dict, ro_nsr_public_key: dict, cloud_config: dict
1341     ) -> None:
1342         """Add ssh keys to cloud config.
1343
1344         Args:
1345            target_vdu  (dict):                 Details of VDU to be created
1346            ro_nsr_public_key   (dict):          RO NSR public Key
1347            cloud_config  (dict):               Cloud config details
1348
1349         """
1350         ssh_keys = []
1351
1352         if target_vdu.get("ssh-keys"):
1353             ssh_keys += target_vdu.get("ssh-keys")
1354
1355         if target_vdu.get("ssh-access-required"):
1356             ssh_keys.append(ro_nsr_public_key)
1357
1358         if ssh_keys:
1359             cloud_config["key-pairs"] = ssh_keys
1360
1361     @staticmethod
1362     def _select_persistent_root_disk(vsd: dict, vdu: dict) -> dict:
1363         """Selects the persistent root disk if exists.
1364         Args:
1365             vsd (dict):             Virtual storage descriptors in VNFD
1366             vdu (dict):             VNF descriptor
1367
1368         Returns:
1369             root_disk   (dict):     Selected persistent root disk
1370         """
1371         if vsd.get("id") == vdu.get("virtual-storage-desc", [[]])[0]:
1372             root_disk = vsd
1373             if root_disk.get(
1374                 "type-of-storage"
1375             ) == "persistent-storage:persistent-storage" and root_disk.get(
1376                 "size-of-storage"
1377             ):
1378                 return root_disk
1379
1380     @staticmethod
1381     def _add_persistent_root_disk_to_disk_list(
1382         vnfd: dict, target_vdu: dict, persistent_root_disk: dict, disk_list: list
1383     ) -> None:
1384         """Find the persistent root disk and add to disk list.
1385
1386         Args:
1387             vnfd  (dict):                           VNF descriptor
1388             target_vdu  (dict):                     Details of VDU to be created
1389             persistent_root_disk    (dict):         Details of persistent root disk
1390             disk_list   (list):                     Disks of VDU
1391
1392         """
1393         for vdu in vnfd.get("vdu", ()):
1394             if vdu["name"] == target_vdu["vdu-name"]:
1395                 for vsd in vnfd.get("virtual-storage-desc", ()):
1396                     root_disk = Ns._select_persistent_root_disk(vsd, vdu)
1397
1398                     if not root_disk:
1399                         continue
1400
1401                     persistent_root_disk[vsd["id"]] = {
1402                         "image_id": vdu.get("sw-image-desc"),
1403                         "size": root_disk["size-of-storage"],
1404                     }
1405
1406                     disk_list.append(persistent_root_disk[vsd["id"]])
1407                     break
1408
1409     @staticmethod
1410     def _add_persistent_ordinary_disks_to_disk_list(
1411         target_vdu: dict,
1412         persistent_root_disk: dict,
1413         persistent_ordinary_disk: dict,
1414         disk_list: list,
1415     ) -> None:
1416         """Fill the disk list by adding persistent ordinary disks.
1417
1418         Args:
1419             target_vdu  (dict):                     Details of VDU to be created
1420             persistent_root_disk    (dict):         Details of persistent root disk
1421             persistent_ordinary_disk    (dict):     Details of persistent ordinary disk
1422             disk_list   (list):                     Disks of VDU
1423
1424         """
1425         if target_vdu.get("virtual-storages"):
1426             for disk in target_vdu["virtual-storages"]:
1427                 if (
1428                     disk.get("type-of-storage")
1429                     == "persistent-storage:persistent-storage"
1430                     and disk["id"] not in persistent_root_disk.keys()
1431                 ):
1432                     persistent_ordinary_disk[disk["id"]] = {
1433                         "size": disk["size-of-storage"],
1434                     }
1435                     disk_list.append(persistent_ordinary_disk[disk["id"]])
1436
1437     @staticmethod
1438     def _prepare_vdu_affinity_group_list(
1439         target_vdu: dict, extra_dict: dict, ns_preffix: str
1440     ) -> List[Dict[str, any]]:
1441         """Process affinity group details to prepare affinity group list.
1442
1443         Args:
1444             target_vdu  (dict):     Details of VDU to be created
1445             extra_dict  (dict):     Dictionary to be filled
1446             ns_preffix  (str):      Prefix as string
1447
1448         Returns:
1449
1450             affinity_group_list (list):     Affinity group details
1451
1452         """
1453         affinity_group_list = []
1454
1455         if target_vdu.get("affinity-or-anti-affinity-group-id"):
1456             for affinity_group_id in target_vdu["affinity-or-anti-affinity-group-id"]:
1457                 affinity_group = {}
1458                 affinity_group_text = (
1459                     ns_preffix + ":affinity-or-anti-affinity-group." + affinity_group_id
1460                 )
1461
1462                 if not isinstance(extra_dict.get("depends_on"), list):
1463                     raise NsException("Invalid extra_dict format.")
1464
1465                 extra_dict["depends_on"].append(affinity_group_text)
1466                 affinity_group["affinity_group_id"] = "TASK-" + affinity_group_text
1467                 affinity_group_list.append(affinity_group)
1468
1469         return affinity_group_list
1470
1471     @staticmethod
1472     def _process_vdu_params(
1473         target_vdu: Dict[str, Any],
1474         indata: Dict[str, Any],
1475         vim_info: Dict[str, Any],
1476         target_record_id: str,
1477         **kwargs: Dict[str, Any],
1478     ) -> Dict[str, Any]:
1479         """Function to process VDU parameters.
1480
1481         Args:
1482             target_vdu (Dict[str, Any]): [description]
1483             indata (Dict[str, Any]): [description]
1484             vim_info (Dict[str, Any]): [description]
1485             target_record_id (str): [description]
1486
1487         Returns:
1488             Dict[str, Any]: [description]
1489         """
1490         vnfr_id = kwargs.get("vnfr_id")
1491         nsr_id = kwargs.get("nsr_id")
1492         vnfr = kwargs.get("vnfr")
1493         vdu2cloud_init = kwargs.get("vdu2cloud_init")
1494         tasks_by_target_record_id = kwargs.get("tasks_by_target_record_id")
1495         logger = kwargs.get("logger")
1496         db = kwargs.get("db")
1497         fs = kwargs.get("fs")
1498         ro_nsr_public_key = kwargs.get("ro_nsr_public_key")
1499
1500         vnf_preffix = "vnfrs:{}".format(vnfr_id)
1501         ns_preffix = "nsrs:{}".format(nsr_id)
1502         image_text = ns_preffix + ":image." + target_vdu["ns-image-id"]
1503         flavor_text = ns_preffix + ":flavor." + target_vdu["ns-flavor-id"]
1504         extra_dict = {"depends_on": [image_text, flavor_text]}
1505         net_list = []
1506
1507         persistent_root_disk = {}
1508         persistent_ordinary_disk = {}
1509         vdu_instantiation_volumes_list = []
1510         disk_list = []
1511         vnfd_id = vnfr["vnfd-id"]
1512         vnfd = db.get_one("vnfds", {"_id": vnfd_id})
1513
1514         # If the position info is provided for all the interfaces, it will be sorted
1515         # according to position number ascendingly.
1516         if all(
1517             True if i.get("position") is not None else False
1518             for i in target_vdu["interfaces"]
1519         ):
1520             Ns._sort_vdu_interfaces(target_vdu)
1521
1522         # If the position info is provided for some interfaces but not all of them, the interfaces
1523         # which has specific position numbers will be placed and others' positions will not be taken care.
1524         else:
1525             Ns._partially_locate_vdu_interfaces(target_vdu)
1526
1527         # If the position info is not provided for the interfaces, interfaces will be attached
1528         # according to the order in the VNFD.
1529         Ns._prepare_vdu_interfaces(
1530             target_vdu,
1531             extra_dict,
1532             ns_preffix,
1533             vnf_preffix,
1534             logger,
1535             tasks_by_target_record_id,
1536             net_list,
1537         )
1538
1539         # cloud config
1540         cloud_config = Ns._prepare_vdu_cloud_init(target_vdu, vdu2cloud_init, db, fs)
1541
1542         # Prepare VDU ssh keys
1543         Ns._prepare_vdu_ssh_keys(target_vdu, ro_nsr_public_key, cloud_config)
1544
1545         if target_vdu.get("additionalParams"):
1546             vdu_instantiation_volumes_list = (
1547                 target_vdu.get("additionalParams").get("OSM").get("vdu_volumes")
1548             )
1549
1550         if vdu_instantiation_volumes_list:
1551             # Find the root volumes and add to the disk_list
1552             persistent_root_disk = Ns.find_persistent_root_volumes(
1553                 vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
1554             )
1555
1556             # Find the ordinary volumes which are not added to the persistent_root_disk
1557             # and put them to the disk list
1558             Ns.find_persistent_volumes(
1559                 persistent_root_disk,
1560                 target_vdu,
1561                 vdu_instantiation_volumes_list,
1562                 disk_list,
1563             )
1564
1565         else:
1566             # Vdu_instantiation_volumes_list is empty
1567             # First get add the persistent root disks to disk_list
1568             Ns._add_persistent_root_disk_to_disk_list(
1569                 vnfd, target_vdu, persistent_root_disk, disk_list
1570             )
1571             # Add the persistent non-root disks to disk_list
1572             Ns._add_persistent_ordinary_disks_to_disk_list(
1573                 target_vdu, persistent_root_disk, persistent_ordinary_disk, disk_list
1574             )
1575
1576         affinity_group_list = Ns._prepare_vdu_affinity_group_list(
1577             target_vdu, extra_dict, ns_preffix
1578         )
1579
1580         extra_dict["params"] = {
1581             "name": "{}-{}-{}-{}".format(
1582                 indata["name"][:16],
1583                 vnfr["member-vnf-index-ref"][:16],
1584                 target_vdu["vdu-name"][:32],
1585                 target_vdu.get("count-index") or 0,
1586             ),
1587             "description": target_vdu["vdu-name"],
1588             "start": True,
1589             "image_id": "TASK-" + image_text,
1590             "flavor_id": "TASK-" + flavor_text,
1591             "affinity_group_list": affinity_group_list,
1592             "net_list": net_list,
1593             "cloud_config": cloud_config or None,
1594             "disk_list": disk_list,
1595             "availability_zone_index": None,  # TODO
1596             "availability_zone_list": None,  # TODO
1597         }
1598
1599         return extra_dict
1600
1601     @staticmethod
1602     def _process_affinity_group_params(
1603         target_affinity_group: Dict[str, Any],
1604         indata: Dict[str, Any],
1605         vim_info: Dict[str, Any],
1606         target_record_id: str,
1607         **kwargs: Dict[str, Any],
1608     ) -> Dict[str, Any]:
1609         """Get affinity or anti-affinity group parameters.
1610
1611         Args:
1612             target_affinity_group (Dict[str, Any]): [description]
1613             indata (Dict[str, Any]): [description]
1614             vim_info (Dict[str, Any]): [description]
1615             target_record_id (str): [description]
1616
1617         Returns:
1618             Dict[str, Any]: [description]
1619         """
1620
1621         extra_dict = {}
1622         affinity_group_data = {
1623             "name": target_affinity_group["name"],
1624             "type": target_affinity_group["type"],
1625             "scope": target_affinity_group["scope"],
1626         }
1627
1628         if target_affinity_group.get("vim-affinity-group-id"):
1629             affinity_group_data["vim-affinity-group-id"] = target_affinity_group[
1630                 "vim-affinity-group-id"
1631             ]
1632
1633         extra_dict["params"] = {
1634             "affinity_group_data": affinity_group_data,
1635         }
1636
1637         return extra_dict
1638
1639     @staticmethod
1640     def _process_recreate_vdu_params(
1641         existing_vdu: Dict[str, Any],
1642         db_nsr: Dict[str, Any],
1643         vim_info: Dict[str, Any],
1644         target_record_id: str,
1645         target_id: str,
1646         **kwargs: Dict[str, Any],
1647     ) -> Dict[str, Any]:
1648         """Function to process VDU parameters to recreate.
1649
1650         Args:
1651             existing_vdu (Dict[str, Any]): [description]
1652             db_nsr (Dict[str, Any]): [description]
1653             vim_info (Dict[str, Any]): [description]
1654             target_record_id (str): [description]
1655             target_id (str): [description]
1656
1657         Returns:
1658             Dict[str, Any]: [description]
1659         """
1660         vnfr = kwargs.get("vnfr")
1661         vdu2cloud_init = kwargs.get("vdu2cloud_init")
1662         # logger = kwargs.get("logger")
1663         db = kwargs.get("db")
1664         fs = kwargs.get("fs")
1665         ro_nsr_public_key = kwargs.get("ro_nsr_public_key")
1666
1667         extra_dict = {}
1668         net_list = []
1669
1670         vim_details = {}
1671         vim_details_text = existing_vdu["vim_info"][target_id].get("vim_details", None)
1672         if vim_details_text:
1673             vim_details = yaml.safe_load(f"{vim_details_text}")
1674
1675         for iface_index, interface in enumerate(existing_vdu["interfaces"]):
1676             if "port-security-enabled" in interface:
1677                 interface["port_security"] = interface.pop("port-security-enabled")
1678
1679             if "port-security-disable-strategy" in interface:
1680                 interface["port_security_disable_strategy"] = interface.pop(
1681                     "port-security-disable-strategy"
1682                 )
1683
1684             net_item = {
1685                 x: v
1686                 for x, v in interface.items()
1687                 if x
1688                 in (
1689                     "name",
1690                     "vpci",
1691                     "port_security",
1692                     "port_security_disable_strategy",
1693                     "floating_ip",
1694                 )
1695             }
1696             existing_ifaces = existing_vdu["vim_info"][target_id].get(
1697                 "interfaces_backup", []
1698             )
1699             net_id = next(
1700                 (
1701                     i["vim_net_id"]
1702                     for i in existing_ifaces
1703                     if i["ip_address"] == interface["ip-address"]
1704                 ),
1705                 None,
1706             )
1707
1708             net_item["net_id"] = net_id
1709             net_item["type"] = "virtual"
1710
1711             # TODO mac_address: used for  SR-IOV ifaces #TODO for other types
1712             # TODO floating_ip: True/False (or it can be None)
1713             if interface.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
1714                 net_item["use"] = "data"
1715                 net_item["model"] = interface["type"]
1716                 net_item["type"] = interface["type"]
1717             elif (
1718                 interface.get("type") == "OM-MGMT"
1719                 or interface.get("mgmt-interface")
1720                 or interface.get("mgmt-vnf")
1721             ):
1722                 net_item["use"] = "mgmt"
1723             else:
1724                 # if interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"):
1725                 net_item["use"] = "bridge"
1726                 net_item["model"] = interface.get("type")
1727
1728             if interface.get("ip-address"):
1729                 net_item["ip_address"] = interface["ip-address"]
1730
1731             if interface.get("mac-address"):
1732                 net_item["mac_address"] = interface["mac-address"]
1733
1734             net_list.append(net_item)
1735
1736             if interface.get("mgmt-vnf"):
1737                 extra_dict["mgmt_vnf_interface"] = iface_index
1738             elif interface.get("mgmt-interface"):
1739                 extra_dict["mgmt_vdu_interface"] = iface_index
1740
1741         # cloud config
1742         cloud_config = {}
1743
1744         if existing_vdu.get("cloud-init"):
1745             if existing_vdu["cloud-init"] not in vdu2cloud_init:
1746                 vdu2cloud_init[existing_vdu["cloud-init"]] = Ns._get_cloud_init(
1747                     db=db,
1748                     fs=fs,
1749                     location=existing_vdu["cloud-init"],
1750                 )
1751
1752             cloud_content_ = vdu2cloud_init[existing_vdu["cloud-init"]]
1753             cloud_config["user-data"] = Ns._parse_jinja2(
1754                 cloud_init_content=cloud_content_,
1755                 params=existing_vdu.get("additionalParams"),
1756                 context=existing_vdu["cloud-init"],
1757             )
1758
1759         if existing_vdu.get("boot-data-drive"):
1760             cloud_config["boot-data-drive"] = existing_vdu.get("boot-data-drive")
1761
1762         ssh_keys = []
1763
1764         if existing_vdu.get("ssh-keys"):
1765             ssh_keys += existing_vdu.get("ssh-keys")
1766
1767         if existing_vdu.get("ssh-access-required"):
1768             ssh_keys.append(ro_nsr_public_key)
1769
1770         if ssh_keys:
1771             cloud_config["key-pairs"] = ssh_keys
1772
1773         disk_list = []
1774         for vol_id in vim_details.get("os-extended-volumes:volumes_attached", []):
1775             disk_list.append({"vim_id": vol_id["id"]})
1776
1777         affinity_group_list = []
1778
1779         if existing_vdu.get("affinity-or-anti-affinity-group-id"):
1780             affinity_group = {}
1781             for affinity_group_id in existing_vdu["affinity-or-anti-affinity-group-id"]:
1782                 for group in db_nsr.get("affinity-or-anti-affinity-group"):
1783                     if (
1784                         group["id"] == affinity_group_id
1785                         and group["vim_info"][target_id].get("vim_id", None) is not None
1786                     ):
1787                         affinity_group["affinity_group_id"] = group["vim_info"][
1788                             target_id
1789                         ].get("vim_id", None)
1790                         affinity_group_list.append(affinity_group)
1791
1792         extra_dict["params"] = {
1793             "name": "{}-{}-{}-{}".format(
1794                 db_nsr["name"][:16],
1795                 vnfr["member-vnf-index-ref"][:16],
1796                 existing_vdu["vdu-name"][:32],
1797                 existing_vdu.get("count-index") or 0,
1798             ),
1799             "description": existing_vdu["vdu-name"],
1800             "start": True,
1801             "image_id": vim_details["image"]["id"],
1802             "flavor_id": vim_details["flavor"]["id"],
1803             "affinity_group_list": affinity_group_list,
1804             "net_list": net_list,
1805             "cloud_config": cloud_config or None,
1806             "disk_list": disk_list,
1807             "availability_zone_index": None,  # TODO
1808             "availability_zone_list": None,  # TODO
1809         }
1810
1811         return extra_dict
1812
1813     def calculate_diff_items(
1814         self,
1815         indata,
1816         db_nsr,
1817         db_ro_nsr,
1818         db_nsr_update,
1819         item,
1820         tasks_by_target_record_id,
1821         action_id,
1822         nsr_id,
1823         task_index,
1824         vnfr_id=None,
1825         vnfr=None,
1826     ):
1827         """Function that returns the incremental changes (creation, deletion)
1828         related to a specific item `item` to be done. This function should be
1829         called for NS instantiation, NS termination, NS update to add a new VNF
1830         or a new VLD, remove a VNF or VLD, etc.
1831         Item can be `net`, `flavor`, `image` or `vdu`.
1832         It takes a list of target items from indata (which came from the REST API)
1833         and compares with the existing items from db_ro_nsr, identifying the
1834         incremental changes to be done. During the comparison, it calls the method
1835         `process_params` (which was passed as parameter, and is particular for each
1836         `item`)
1837
1838         Args:
1839             indata (Dict[str, Any]): deployment info
1840             db_nsr: NSR record from DB
1841             db_ro_nsr (Dict[str, Any]): record from "ro_nsrs"
1842             db_nsr_update (Dict[str, Any]): NSR info to update in DB
1843             item (str): element to process (net, vdu...)
1844             tasks_by_target_record_id (Dict[str, Any]):
1845                 [<target_record_id>, <task>]
1846             action_id (str): action id
1847             nsr_id (str): NSR id
1848             task_index (number): task index to add to task name
1849             vnfr_id (str): VNFR id
1850             vnfr (Dict[str, Any]): VNFR info
1851
1852         Returns:
1853             List: list with the incremental changes (deletes, creates) for each item
1854             number: current task index
1855         """
1856
1857         diff_items = []
1858         db_path = ""
1859         db_record = ""
1860         target_list = []
1861         existing_list = []
1862         process_params = None
1863         vdu2cloud_init = indata.get("cloud_init_content") or {}
1864         ro_nsr_public_key = db_ro_nsr["public_key"]
1865
1866         # According to the type of item, the path, the target_list,
1867         # the existing_list and the method to process params are set
1868         db_path = self.db_path_map[item]
1869         process_params = self.process_params_function_map[item]
1870         if item in ("net", "vdu"):
1871             # This case is specific for the NS VLD (not applied to VDU)
1872             if vnfr is None:
1873                 db_record = "nsrs:{}:{}".format(nsr_id, db_path)
1874                 target_list = indata.get("ns", []).get(db_path, [])
1875                 existing_list = db_nsr.get(db_path, [])
1876             # This case is common for VNF VLDs and VNF VDUs
1877             else:
1878                 db_record = "vnfrs:{}:{}".format(vnfr_id, db_path)
1879                 target_vnf = next(
1880                     (vnf for vnf in indata.get("vnf", ()) if vnf["_id"] == vnfr_id),
1881                     None,
1882                 )
1883                 target_list = target_vnf.get(db_path, []) if target_vnf else []
1884                 existing_list = vnfr.get(db_path, [])
1885         elif item in ("image", "flavor", "affinity-or-anti-affinity-group"):
1886             db_record = "nsrs:{}:{}".format(nsr_id, db_path)
1887             target_list = indata.get(item, [])
1888             existing_list = db_nsr.get(item, [])
1889         else:
1890             raise NsException("Item not supported: {}", item)
1891
1892         # ensure all the target_list elements has an "id". If not assign the index as id
1893         if target_list is None:
1894             target_list = []
1895         for target_index, tl in enumerate(target_list):
1896             if tl and not tl.get("id"):
1897                 tl["id"] = str(target_index)
1898
1899         # step 1 items (networks,vdus,...) to be deleted/updated
1900         for item_index, existing_item in enumerate(existing_list):
1901             target_item = next(
1902                 (t for t in target_list if t["id"] == existing_item["id"]),
1903                 None,
1904             )
1905
1906             for target_vim, existing_viminfo in existing_item.get(
1907                 "vim_info", {}
1908             ).items():
1909                 if existing_viminfo is None:
1910                     continue
1911
1912                 if target_item:
1913                     target_viminfo = target_item.get("vim_info", {}).get(target_vim)
1914                 else:
1915                     target_viminfo = None
1916
1917                 if target_viminfo is None:
1918                     # must be deleted
1919                     self._assign_vim(target_vim)
1920                     target_record_id = "{}.{}".format(db_record, existing_item["id"])
1921                     item_ = item
1922
1923                     if target_vim.startswith("sdn") or target_vim.startswith("wim"):
1924                         # item must be sdn-net instead of net if target_vim is a sdn
1925                         item_ = "sdn_net"
1926                         target_record_id += ".sdn"
1927
1928                     deployment_info = {
1929                         "action_id": action_id,
1930                         "nsr_id": nsr_id,
1931                         "task_index": task_index,
1932                     }
1933
1934                     diff_items.append(
1935                         {
1936                             "deployment_info": deployment_info,
1937                             "target_id": target_vim,
1938                             "item": item_,
1939                             "action": "DELETE",
1940                             "target_record": f"{db_record}.{item_index}.vim_info.{target_vim}",
1941                             "target_record_id": target_record_id,
1942                         }
1943                     )
1944                     task_index += 1
1945
1946         # step 2 items (networks,vdus,...) to be created
1947         for target_item in target_list:
1948             item_index = -1
1949
1950             for item_index, existing_item in enumerate(existing_list):
1951                 if existing_item["id"] == target_item["id"]:
1952                     break
1953             else:
1954                 item_index += 1
1955                 db_nsr_update[db_path + ".{}".format(item_index)] = target_item
1956                 existing_list.append(target_item)
1957                 existing_item = None
1958
1959             for target_vim, target_viminfo in target_item.get("vim_info", {}).items():
1960                 existing_viminfo = None
1961
1962                 if existing_item:
1963                     existing_viminfo = existing_item.get("vim_info", {}).get(target_vim)
1964
1965                 if existing_viminfo is not None:
1966                     continue
1967
1968                 target_record_id = "{}.{}".format(db_record, target_item["id"])
1969                 item_ = item
1970
1971                 if target_vim.startswith("sdn") or target_vim.startswith("wim"):
1972                     # item must be sdn-net instead of net if target_vim is a sdn
1973                     item_ = "sdn_net"
1974                     target_record_id += ".sdn"
1975
1976                 kwargs = {}
1977                 self.logger.warning(
1978                     "ns.calculate_diff_items target_item={}".format(target_item)
1979                 )
1980                 if process_params == Ns._process_flavor_params:
1981                     kwargs.update(
1982                         {
1983                             "db": self.db,
1984                         }
1985                     )
1986                     self.logger.warning(
1987                         "calculate_diff_items for flavor kwargs={}".format(kwargs)
1988                     )
1989
1990                 if process_params == Ns._process_vdu_params:
1991                     self.logger.warning(
1992                         "calculate_diff_items self.fs={}".format(self.fs)
1993                     )
1994                     kwargs.update(
1995                         {
1996                             "vnfr_id": vnfr_id,
1997                             "nsr_id": nsr_id,
1998                             "vnfr": vnfr,
1999                             "vdu2cloud_init": vdu2cloud_init,
2000                             "tasks_by_target_record_id": tasks_by_target_record_id,
2001                             "logger": self.logger,
2002                             "db": self.db,
2003                             "fs": self.fs,
2004                             "ro_nsr_public_key": ro_nsr_public_key,
2005                         }
2006                     )
2007                     self.logger.warning("calculate_diff_items kwargs={}".format(kwargs))
2008
2009                 extra_dict = process_params(
2010                     target_item,
2011                     indata,
2012                     target_viminfo,
2013                     target_record_id,
2014                     **kwargs,
2015                 )
2016                 self._assign_vim(target_vim)
2017
2018                 deployment_info = {
2019                     "action_id": action_id,
2020                     "nsr_id": nsr_id,
2021                     "task_index": task_index,
2022                 }
2023
2024                 new_item = {
2025                     "deployment_info": deployment_info,
2026                     "target_id": target_vim,
2027                     "item": item_,
2028                     "action": "CREATE",
2029                     "target_record": f"{db_record}.{item_index}.vim_info.{target_vim}",
2030                     "target_record_id": target_record_id,
2031                     "extra_dict": extra_dict,
2032                     "common_id": target_item.get("common_id", None),
2033                 }
2034                 diff_items.append(new_item)
2035                 tasks_by_target_record_id[target_record_id] = new_item
2036                 task_index += 1
2037
2038                 db_nsr_update[db_path + ".{}".format(item_index)] = target_item
2039
2040         return diff_items, task_index
2041
2042     def calculate_all_differences_to_deploy(
2043         self,
2044         indata,
2045         nsr_id,
2046         db_nsr,
2047         db_vnfrs,
2048         db_ro_nsr,
2049         db_nsr_update,
2050         db_vnfrs_update,
2051         action_id,
2052         tasks_by_target_record_id,
2053     ):
2054         """This method calculates the ordered list of items (`changes_list`)
2055         to be created and deleted.
2056
2057         Args:
2058             indata (Dict[str, Any]): deployment info
2059             nsr_id (str): NSR id
2060             db_nsr: NSR record from DB
2061             db_vnfrs: VNFRS record from DB
2062             db_ro_nsr (Dict[str, Any]): record from "ro_nsrs"
2063             db_nsr_update (Dict[str, Any]): NSR info to update in DB
2064             db_vnfrs_update (Dict[str, Any]): VNFRS info to update in DB
2065             action_id (str): action id
2066             tasks_by_target_record_id (Dict[str, Any]):
2067                 [<target_record_id>, <task>]
2068
2069         Returns:
2070             List: ordered list of items to be created and deleted.
2071         """
2072
2073         task_index = 0
2074         # set list with diffs:
2075         changes_list = []
2076
2077         # NS vld, image and flavor
2078         for item in ["net", "image", "flavor", "affinity-or-anti-affinity-group"]:
2079             self.logger.debug("process NS={} {}".format(nsr_id, item))
2080             diff_items, task_index = self.calculate_diff_items(
2081                 indata=indata,
2082                 db_nsr=db_nsr,
2083                 db_ro_nsr=db_ro_nsr,
2084                 db_nsr_update=db_nsr_update,
2085                 item=item,
2086                 tasks_by_target_record_id=tasks_by_target_record_id,
2087                 action_id=action_id,
2088                 nsr_id=nsr_id,
2089                 task_index=task_index,
2090                 vnfr_id=None,
2091             )
2092             changes_list += diff_items
2093
2094         # VNF vlds and vdus
2095         for vnfr_id, vnfr in db_vnfrs.items():
2096             # vnfr_id need to be set as global variable for among others nested method _process_vdu_params
2097             for item in ["net", "vdu"]:
2098                 self.logger.debug("process VNF={} {}".format(vnfr_id, item))
2099                 diff_items, task_index = self.calculate_diff_items(
2100                     indata=indata,
2101                     db_nsr=db_nsr,
2102                     db_ro_nsr=db_ro_nsr,
2103                     db_nsr_update=db_vnfrs_update[vnfr["_id"]],
2104                     item=item,
2105                     tasks_by_target_record_id=tasks_by_target_record_id,
2106                     action_id=action_id,
2107                     nsr_id=nsr_id,
2108                     task_index=task_index,
2109                     vnfr_id=vnfr_id,
2110                     vnfr=vnfr,
2111                 )
2112                 changes_list += diff_items
2113
2114         return changes_list
2115
2116     def define_all_tasks(
2117         self,
2118         changes_list,
2119         db_new_tasks,
2120         tasks_by_target_record_id,
2121     ):
2122         """Function to create all the task structures obtanied from
2123         the method calculate_all_differences_to_deploy
2124
2125         Args:
2126             changes_list (List): ordered list of items to be created or deleted
2127             db_new_tasks (List): tasks list to be created
2128             action_id (str): action id
2129             tasks_by_target_record_id (Dict[str, Any]):
2130                 [<target_record_id>, <task>]
2131
2132         """
2133
2134         for change in changes_list:
2135             task = Ns._create_task(
2136                 deployment_info=change["deployment_info"],
2137                 target_id=change["target_id"],
2138                 item=change["item"],
2139                 action=change["action"],
2140                 target_record=change["target_record"],
2141                 target_record_id=change["target_record_id"],
2142                 extra_dict=change.get("extra_dict", None),
2143             )
2144
2145             self.logger.warning("ns.define_all_tasks task={}".format(task))
2146             tasks_by_target_record_id[change["target_record_id"]] = task
2147             db_new_tasks.append(task)
2148
2149             if change.get("common_id"):
2150                 task["common_id"] = change["common_id"]
2151
2152     def upload_all_tasks(
2153         self,
2154         db_new_tasks,
2155         now,
2156     ):
2157         """Function to save all tasks in the common DB
2158
2159         Args:
2160             db_new_tasks (List): tasks list to be created
2161             now (time): current time
2162
2163         """
2164
2165         nb_ro_tasks = 0  # for logging
2166
2167         for db_task in db_new_tasks:
2168             target_id = db_task.pop("target_id")
2169             common_id = db_task.get("common_id")
2170
2171             # Do not chek tasks with vim_status DELETED
2172             # because in manual heealing there are two tasks for the same vdur:
2173             #   one with vim_status deleted and the other one with the actual VM status.
2174
2175             if common_id:
2176                 if self.db.set_one(
2177                     "ro_tasks",
2178                     q_filter={
2179                         "target_id": target_id,
2180                         "tasks.common_id": common_id,
2181                         "vim_info.vim_status.ne": "DELETED",
2182                     },
2183                     update_dict={"to_check_at": now, "modified_at": now},
2184                     push={"tasks": db_task},
2185                     fail_on_empty=False,
2186                 ):
2187                     continue
2188
2189             if not self.db.set_one(
2190                 "ro_tasks",
2191                 q_filter={
2192                     "target_id": target_id,
2193                     "tasks.target_record": db_task["target_record"],
2194                     "vim_info.vim_status.ne": "DELETED",
2195                 },
2196                 update_dict={"to_check_at": now, "modified_at": now},
2197                 push={"tasks": db_task},
2198                 fail_on_empty=False,
2199             ):
2200                 # Create a ro_task
2201                 self.logger.debug("Updating database, Creating ro_tasks")
2202                 db_ro_task = Ns._create_ro_task(target_id, db_task)
2203                 nb_ro_tasks += 1
2204                 self.db.create("ro_tasks", db_ro_task)
2205
2206         self.logger.debug(
2207             "Created {} ro_tasks; {} tasks - db_new_tasks={}".format(
2208                 nb_ro_tasks, len(db_new_tasks), db_new_tasks
2209             )
2210         )
2211
2212     def upload_recreate_tasks(
2213         self,
2214         db_new_tasks,
2215         now,
2216     ):
2217         """Function to save recreate tasks in the common DB
2218
2219         Args:
2220             db_new_tasks (List): tasks list to be created
2221             now (time): current time
2222
2223         """
2224
2225         nb_ro_tasks = 0  # for logging
2226
2227         for db_task in db_new_tasks:
2228             target_id = db_task.pop("target_id")
2229             self.logger.warning("target_id={} db_task={}".format(target_id, db_task))
2230
2231             action = db_task.get("action", None)
2232
2233             # Create a ro_task
2234             self.logger.debug("Updating database, Creating ro_tasks")
2235             db_ro_task = Ns._create_ro_task(target_id, db_task)
2236
2237             # If DELETE task: the associated created items should be removed
2238             # (except persistent volumes):
2239             if action == "DELETE":
2240                 db_ro_task["vim_info"]["created"] = True
2241                 db_ro_task["vim_info"]["created_items"] = db_task.get(
2242                     "created_items", {}
2243                 )
2244                 db_ro_task["vim_info"]["volumes_to_hold"] = db_task.get(
2245                     "volumes_to_hold", []
2246                 )
2247                 db_ro_task["vim_info"]["vim_id"] = db_task.get("vim_id", None)
2248
2249             nb_ro_tasks += 1
2250             self.logger.warning("upload_all_tasks db_ro_task={}".format(db_ro_task))
2251             self.db.create("ro_tasks", db_ro_task)
2252
2253         self.logger.debug(
2254             "Created {} ro_tasks; {} tasks - db_new_tasks={}".format(
2255                 nb_ro_tasks, len(db_new_tasks), db_new_tasks
2256             )
2257         )
2258
2259     def _prepare_created_items_for_healing(
2260         self,
2261         nsr_id,
2262         target_record,
2263     ):
2264         created_items = {}
2265         # Get created_items from ro_task
2266         ro_tasks = self.db.get_list("ro_tasks", {"tasks.nsr_id": nsr_id})
2267         for ro_task in ro_tasks:
2268             for task in ro_task["tasks"]:
2269                 if (
2270                     task["target_record"] == target_record
2271                     and task["action"] == "CREATE"
2272                     and ro_task["vim_info"]["created_items"]
2273                 ):
2274                     created_items = ro_task["vim_info"]["created_items"]
2275                     break
2276
2277         return created_items
2278
2279     def _prepare_persistent_volumes_for_healing(
2280         self,
2281         target_id,
2282         existing_vdu,
2283     ):
2284         # The associated volumes of the VM shouldn't be removed
2285         volumes_list = []
2286         vim_details = {}
2287         vim_details_text = existing_vdu["vim_info"][target_id].get("vim_details", None)
2288         if vim_details_text:
2289             vim_details = yaml.safe_load(f"{vim_details_text}")
2290
2291             for vol_id in vim_details.get("os-extended-volumes:volumes_attached", []):
2292                 volumes_list.append(vol_id["id"])
2293
2294         return volumes_list
2295
2296     def prepare_changes_to_recreate(
2297         self,
2298         indata,
2299         nsr_id,
2300         db_nsr,
2301         db_vnfrs,
2302         db_ro_nsr,
2303         action_id,
2304         tasks_by_target_record_id,
2305     ):
2306         """This method will obtain an ordered list of items (`changes_list`)
2307         to be created and deleted to meet the recreate request.
2308         """
2309
2310         self.logger.debug(
2311             "ns.prepare_changes_to_recreate nsr_id={} indata={}".format(nsr_id, indata)
2312         )
2313
2314         task_index = 0
2315         # set list with diffs:
2316         changes_list = []
2317         db_path = self.db_path_map["vdu"]
2318         target_list = indata.get("healVnfData", {})
2319         vdu2cloud_init = indata.get("cloud_init_content") or {}
2320         ro_nsr_public_key = db_ro_nsr["public_key"]
2321
2322         # Check each VNF of the target
2323         for target_vnf in target_list:
2324             # Find this VNF in the list from DB, raise exception if vnfInstanceId is not found
2325             vnfr_id = target_vnf["vnfInstanceId"]
2326             existing_vnf = db_vnfrs.get(vnfr_id)
2327             db_record = "vnfrs:{}:{}".format(vnfr_id, db_path)
2328             # vim_account_id = existing_vnf.get("vim-account-id", "")
2329
2330             target_vdus = target_vnf.get("additionalParams", {}).get("vdu", [])
2331             # Check each VDU of this VNF
2332             if not target_vdus:
2333                 # Create target_vdu_list from DB, if VDUs are not specified
2334                 target_vdus = []
2335                 for existing_vdu in existing_vnf.get("vdur"):
2336                     vdu_name = existing_vdu.get("vdu-name", None)
2337                     vdu_index = existing_vdu.get("count-index", 0)
2338                     vdu_to_be_healed = {"vdu-id": vdu_name, "count-index": vdu_index}
2339                     target_vdus.append(vdu_to_be_healed)
2340             for target_vdu in target_vdus:
2341                 vdu_name = target_vdu.get("vdu-id", None)
2342                 # For multi instance VDU count-index is mandatory
2343                 # For single session VDU count-indes is 0
2344                 count_index = target_vdu.get("count-index", 0)
2345                 item_index = 0
2346                 existing_instance = None
2347                 for instance in existing_vnf.get("vdur", None):
2348                     if (
2349                         instance["vdu-name"] == vdu_name
2350                         and instance["count-index"] == count_index
2351                     ):
2352                         existing_instance = instance
2353                         break
2354                     else:
2355                         item_index += 1
2356
2357                 target_record_id = "{}.{}".format(db_record, existing_instance["id"])
2358
2359                 # The target VIM is the one already existing in DB to recreate
2360                 for target_vim, target_viminfo in existing_instance.get(
2361                     "vim_info", {}
2362                 ).items():
2363                     # step 1 vdu to be deleted
2364                     self._assign_vim(target_vim)
2365                     deployment_info = {
2366                         "action_id": action_id,
2367                         "nsr_id": nsr_id,
2368                         "task_index": task_index,
2369                     }
2370
2371                     target_record = f"{db_record}.{item_index}.vim_info.{target_vim}"
2372                     created_items = self._prepare_created_items_for_healing(
2373                         nsr_id, target_record
2374                     )
2375
2376                     volumes_to_hold = self._prepare_persistent_volumes_for_healing(
2377                         target_vim, existing_instance
2378                     )
2379
2380                     # Specific extra params for recreate tasks:
2381                     extra_dict = {
2382                         "created_items": created_items,
2383                         "vim_id": existing_instance["vim-id"],
2384                         "volumes_to_hold": volumes_to_hold,
2385                     }
2386
2387                     changes_list.append(
2388                         {
2389                             "deployment_info": deployment_info,
2390                             "target_id": target_vim,
2391                             "item": "vdu",
2392                             "action": "DELETE",
2393                             "target_record": target_record,
2394                             "target_record_id": target_record_id,
2395                             "extra_dict": extra_dict,
2396                         }
2397                     )
2398                     delete_task_id = f"{action_id}:{task_index}"
2399                     task_index += 1
2400
2401                     # step 2 vdu to be created
2402                     kwargs = {}
2403                     kwargs.update(
2404                         {
2405                             "vnfr_id": vnfr_id,
2406                             "nsr_id": nsr_id,
2407                             "vnfr": existing_vnf,
2408                             "vdu2cloud_init": vdu2cloud_init,
2409                             "tasks_by_target_record_id": tasks_by_target_record_id,
2410                             "logger": self.logger,
2411                             "db": self.db,
2412                             "fs": self.fs,
2413                             "ro_nsr_public_key": ro_nsr_public_key,
2414                         }
2415                     )
2416
2417                     extra_dict = self._process_recreate_vdu_params(
2418                         existing_instance,
2419                         db_nsr,
2420                         target_viminfo,
2421                         target_record_id,
2422                         target_vim,
2423                         **kwargs,
2424                     )
2425
2426                     # The CREATE task depens on the DELETE task
2427                     extra_dict["depends_on"] = [delete_task_id]
2428
2429                     # Add volumes created from created_items if any
2430                     # Ports should be deleted with delete task and automatically created with create task
2431                     volumes = {}
2432                     for k, v in created_items.items():
2433                         try:
2434                             k_item, _, k_id = k.partition(":")
2435                             if k_item == "volume":
2436                                 volumes[k] = v
2437                         except Exception as e:
2438                             self.logger.error(
2439                                 "Error evaluating created item {}: {}".format(k, e)
2440                             )
2441                     extra_dict["previous_created_volumes"] = volumes
2442
2443                     deployment_info = {
2444                         "action_id": action_id,
2445                         "nsr_id": nsr_id,
2446                         "task_index": task_index,
2447                     }
2448                     self._assign_vim(target_vim)
2449
2450                     new_item = {
2451                         "deployment_info": deployment_info,
2452                         "target_id": target_vim,
2453                         "item": "vdu",
2454                         "action": "CREATE",
2455                         "target_record": target_record,
2456                         "target_record_id": target_record_id,
2457                         "extra_dict": extra_dict,
2458                     }
2459                     changes_list.append(new_item)
2460                     tasks_by_target_record_id[target_record_id] = new_item
2461                     task_index += 1
2462
2463         return changes_list
2464
2465     def recreate(self, session, indata, version, nsr_id, *args, **kwargs):
2466         self.logger.debug("ns.recreate nsr_id={} indata={}".format(nsr_id, indata))
2467         # TODO: validate_input(indata, recreate_schema)
2468         action_id = indata.get("action_id", str(uuid4()))
2469         # get current deployment
2470         db_vnfrs = {}  # vnf's info indexed by _id
2471         step = ""
2472         logging_text = "Recreate nsr_id={} action_id={} indata={}".format(
2473             nsr_id, action_id, indata
2474         )
2475         self.logger.debug(logging_text + "Enter")
2476
2477         try:
2478             step = "Getting ns and vnfr record from db"
2479             db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
2480             db_new_tasks = []
2481             tasks_by_target_record_id = {}
2482             # read from db: vnf's of this ns
2483             step = "Getting vnfrs from db"
2484             db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
2485             self.logger.debug("ns.recreate: db_vnfrs_list={}".format(db_vnfrs_list))
2486
2487             if not db_vnfrs_list:
2488                 raise NsException("Cannot obtain associated VNF for ns")
2489
2490             for vnfr in db_vnfrs_list:
2491                 db_vnfrs[vnfr["_id"]] = vnfr
2492
2493             now = time()
2494             db_ro_nsr = self.db.get_one("ro_nsrs", {"_id": nsr_id}, fail_on_empty=False)
2495             self.logger.debug("ns.recreate: db_ro_nsr={}".format(db_ro_nsr))
2496
2497             if not db_ro_nsr:
2498                 db_ro_nsr = self._create_db_ro_nsrs(nsr_id, now)
2499
2500             with self.write_lock:
2501                 # NS
2502                 step = "process NS elements"
2503                 changes_list = self.prepare_changes_to_recreate(
2504                     indata=indata,
2505                     nsr_id=nsr_id,
2506                     db_nsr=db_nsr,
2507                     db_vnfrs=db_vnfrs,
2508                     db_ro_nsr=db_ro_nsr,
2509                     action_id=action_id,
2510                     tasks_by_target_record_id=tasks_by_target_record_id,
2511                 )
2512
2513                 self.define_all_tasks(
2514                     changes_list=changes_list,
2515                     db_new_tasks=db_new_tasks,
2516                     tasks_by_target_record_id=tasks_by_target_record_id,
2517                 )
2518
2519                 # Delete all ro_tasks registered for the targets vdurs (target_record)
2520                 # If task of type CREATE exist then vim will try to get info form deleted VMs.
2521                 # So remove all task related to target record.
2522                 ro_tasks = self.db.get_list("ro_tasks", {"tasks.nsr_id": nsr_id})
2523                 for change in changes_list:
2524                     for ro_task in ro_tasks:
2525                         for task in ro_task["tasks"]:
2526                             if task["target_record"] == change["target_record"]:
2527                                 self.db.del_one(
2528                                     "ro_tasks",
2529                                     q_filter={
2530                                         "_id": ro_task["_id"],
2531                                         "modified_at": ro_task["modified_at"],
2532                                     },
2533                                     fail_on_empty=False,
2534                                 )
2535
2536                 step = "Updating database, Appending tasks to ro_tasks"
2537                 self.upload_recreate_tasks(
2538                     db_new_tasks=db_new_tasks,
2539                     now=now,
2540                 )
2541
2542             self.logger.debug(
2543                 logging_text + "Exit. Created {} tasks".format(len(db_new_tasks))
2544             )
2545
2546             return (
2547                 {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
2548                 action_id,
2549                 True,
2550             )
2551         except Exception as e:
2552             if isinstance(e, (DbException, NsException)):
2553                 self.logger.error(
2554                     logging_text + "Exit Exception while '{}': {}".format(step, e)
2555                 )
2556             else:
2557                 e = traceback_format_exc()
2558                 self.logger.critical(
2559                     logging_text + "Exit Exception while '{}': {}".format(step, e),
2560                     exc_info=True,
2561                 )
2562
2563             raise NsException(e)
2564
2565     def deploy(self, session, indata, version, nsr_id, *args, **kwargs):
2566         self.logger.debug("ns.deploy nsr_id={} indata={}".format(nsr_id, indata))
2567         validate_input(indata, deploy_schema)
2568         action_id = indata.get("action_id", str(uuid4()))
2569         task_index = 0
2570         # get current deployment
2571         db_nsr_update = {}  # update operation on nsrs
2572         db_vnfrs_update = {}
2573         db_vnfrs = {}  # vnf's info indexed by _id
2574         step = ""
2575         logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id)
2576         self.logger.debug(logging_text + "Enter")
2577
2578         try:
2579             step = "Getting ns and vnfr record from db"
2580             db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
2581             self.logger.debug("ns.deploy: db_nsr={}".format(db_nsr))
2582             db_new_tasks = []
2583             tasks_by_target_record_id = {}
2584             # read from db: vnf's of this ns
2585             step = "Getting vnfrs from db"
2586             db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
2587
2588             if not db_vnfrs_list:
2589                 raise NsException("Cannot obtain associated VNF for ns")
2590
2591             for vnfr in db_vnfrs_list:
2592                 db_vnfrs[vnfr["_id"]] = vnfr
2593                 db_vnfrs_update[vnfr["_id"]] = {}
2594             self.logger.debug("ns.deploy db_vnfrs={}".format(db_vnfrs))
2595
2596             now = time()
2597             db_ro_nsr = self.db.get_one("ro_nsrs", {"_id": nsr_id}, fail_on_empty=False)
2598
2599             if not db_ro_nsr:
2600                 db_ro_nsr = self._create_db_ro_nsrs(nsr_id, now)
2601
2602             # check that action_id is not in the list of actions. Suffixed with :index
2603             if action_id in db_ro_nsr["actions"]:
2604                 index = 1
2605
2606                 while True:
2607                     new_action_id = "{}:{}".format(action_id, index)
2608
2609                     if new_action_id not in db_ro_nsr["actions"]:
2610                         action_id = new_action_id
2611                         self.logger.debug(
2612                             logging_text
2613                             + "Changing action_id in use to {}".format(action_id)
2614                         )
2615                         break
2616
2617                     index += 1
2618
2619             def _process_action(indata):
2620                 nonlocal db_new_tasks
2621                 nonlocal action_id
2622                 nonlocal nsr_id
2623                 nonlocal task_index
2624                 nonlocal db_vnfrs
2625                 nonlocal db_ro_nsr
2626
2627                 if indata["action"]["action"] == "inject_ssh_key":
2628                     key = indata["action"].get("key")
2629                     user = indata["action"].get("user")
2630                     password = indata["action"].get("password")
2631
2632                     for vnf in indata.get("vnf", ()):
2633                         if vnf["_id"] not in db_vnfrs:
2634                             raise NsException("Invalid vnf={}".format(vnf["_id"]))
2635
2636                         db_vnfr = db_vnfrs[vnf["_id"]]
2637
2638                         for target_vdu in vnf.get("vdur", ()):
2639                             vdu_index, vdur = next(
2640                                 (
2641                                     i_v
2642                                     for i_v in enumerate(db_vnfr["vdur"])
2643                                     if i_v[1]["id"] == target_vdu["id"]
2644                                 ),
2645                                 (None, None),
2646                             )
2647
2648                             if not vdur:
2649                                 raise NsException(
2650                                     "Invalid vdu vnf={}.{}".format(
2651                                         vnf["_id"], target_vdu["id"]
2652                                     )
2653                                 )
2654
2655                             target_vim, vim_info = next(
2656                                 k_v for k_v in vdur["vim_info"].items()
2657                             )
2658                             self._assign_vim(target_vim)
2659                             target_record = "vnfrs:{}:vdur.{}.ssh_keys".format(
2660                                 vnf["_id"], vdu_index
2661                             )
2662                             extra_dict = {
2663                                 "depends_on": [
2664                                     "vnfrs:{}:vdur.{}".format(vnf["_id"], vdur["id"])
2665                                 ],
2666                                 "params": {
2667                                     "ip_address": vdur.get("ip-address"),
2668                                     "user": user,
2669                                     "key": key,
2670                                     "password": password,
2671                                     "private_key": db_ro_nsr["private_key"],
2672                                     "salt": db_ro_nsr["_id"],
2673                                     "schema_version": db_ro_nsr["_admin"][
2674                                         "schema_version"
2675                                     ],
2676                                 },
2677                             }
2678
2679                             deployment_info = {
2680                                 "action_id": action_id,
2681                                 "nsr_id": nsr_id,
2682                                 "task_index": task_index,
2683                             }
2684
2685                             task = Ns._create_task(
2686                                 deployment_info=deployment_info,
2687                                 target_id=target_vim,
2688                                 item="vdu",
2689                                 action="EXEC",
2690                                 target_record=target_record,
2691                                 target_record_id=None,
2692                                 extra_dict=extra_dict,
2693                             )
2694
2695                             task_index = deployment_info.get("task_index")
2696
2697                             db_new_tasks.append(task)
2698
2699             with self.write_lock:
2700                 if indata.get("action"):
2701                     _process_action(indata)
2702                 else:
2703                     # compute network differences
2704                     # NS
2705                     step = "process NS elements"
2706                     changes_list = self.calculate_all_differences_to_deploy(
2707                         indata=indata,
2708                         nsr_id=nsr_id,
2709                         db_nsr=db_nsr,
2710                         db_vnfrs=db_vnfrs,
2711                         db_ro_nsr=db_ro_nsr,
2712                         db_nsr_update=db_nsr_update,
2713                         db_vnfrs_update=db_vnfrs_update,
2714                         action_id=action_id,
2715                         tasks_by_target_record_id=tasks_by_target_record_id,
2716                     )
2717                     self.define_all_tasks(
2718                         changes_list=changes_list,
2719                         db_new_tasks=db_new_tasks,
2720                         tasks_by_target_record_id=tasks_by_target_record_id,
2721                     )
2722
2723                 step = "Updating database, Appending tasks to ro_tasks"
2724                 self.upload_all_tasks(
2725                     db_new_tasks=db_new_tasks,
2726                     now=now,
2727                 )
2728
2729                 step = "Updating database, nsrs"
2730                 if db_nsr_update:
2731                     self.db.set_one("nsrs", {"_id": nsr_id}, db_nsr_update)
2732
2733                 for vnfr_id, db_vnfr_update in db_vnfrs_update.items():
2734                     if db_vnfr_update:
2735                         step = "Updating database, vnfrs={}".format(vnfr_id)
2736                         self.db.set_one("vnfrs", {"_id": vnfr_id}, db_vnfr_update)
2737
2738             self.logger.debug(
2739                 logging_text + "Exit. Created {} tasks".format(len(db_new_tasks))
2740             )
2741
2742             return (
2743                 {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
2744                 action_id,
2745                 True,
2746             )
2747         except Exception as e:
2748             if isinstance(e, (DbException, NsException)):
2749                 self.logger.error(
2750                     logging_text + "Exit Exception while '{}': {}".format(step, e)
2751                 )
2752             else:
2753                 e = traceback_format_exc()
2754                 self.logger.critical(
2755                     logging_text + "Exit Exception while '{}': {}".format(step, e),
2756                     exc_info=True,
2757                 )
2758
2759             raise NsException(e)
2760
2761     def delete(self, session, indata, version, nsr_id, *args, **kwargs):
2762         self.logger.debug("ns.delete version={} nsr_id={}".format(version, nsr_id))
2763         # self.db.del_list({"_id": ro_task["_id"], "tasks.nsr_id.ne": nsr_id})
2764
2765         with self.write_lock:
2766             try:
2767                 NsWorker.delete_db_tasks(self.db, nsr_id, None)
2768             except NsWorkerException as e:
2769                 raise NsException(e)
2770
2771         return None, None, True
2772
2773     def status(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
2774         self.logger.debug(
2775             "ns.status version={} nsr_id={}, action_id={} indata={}".format(
2776                 version, nsr_id, action_id, indata
2777             )
2778         )
2779         task_list = []
2780         done = 0
2781         total = 0
2782         ro_tasks = self.db.get_list("ro_tasks", {"tasks.action_id": action_id})
2783         global_status = "DONE"
2784         details = []
2785
2786         for ro_task in ro_tasks:
2787             for task in ro_task["tasks"]:
2788                 if task and task["action_id"] == action_id:
2789                     task_list.append(task)
2790                     total += 1
2791
2792                     if task["status"] == "FAILED":
2793                         global_status = "FAILED"
2794                         error_text = "Error at {} {}: {}".format(
2795                             task["action"].lower(),
2796                             task["item"],
2797                             ro_task["vim_info"].get("vim_message") or "unknown",
2798                         )
2799                         details.append(error_text)
2800                     elif task["status"] in ("SCHEDULED", "BUILD"):
2801                         if global_status != "FAILED":
2802                             global_status = "BUILD"
2803                     else:
2804                         done += 1
2805
2806         return_data = {
2807             "status": global_status,
2808             "details": ". ".join(details)
2809             if details
2810             else "progress {}/{}".format(done, total),
2811             "nsr_id": nsr_id,
2812             "action_id": action_id,
2813             "tasks": task_list,
2814         }
2815
2816         return return_data, None, True
2817
2818     def recreate_status(
2819         self, session, indata, version, nsr_id, action_id, *args, **kwargs
2820     ):
2821         return self.status(session, indata, version, nsr_id, action_id, *args, **kwargs)
2822
2823     def cancel(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
2824         print(
2825             "ns.cancel session={} indata={} version={} nsr_id={}, action_id={}".format(
2826                 session, indata, version, nsr_id, action_id
2827             )
2828         )
2829
2830         return None, None, True
2831
2832     def rebuild_start_stop_task(
2833         self,
2834         vdu_id,
2835         vnf_id,
2836         vdu_index,
2837         action_id,
2838         nsr_id,
2839         task_index,
2840         target_vim,
2841         extra_dict,
2842     ):
2843         self._assign_vim(target_vim)
2844         target_record = "vnfrs:{}:vdur.{}".format(vnf_id, vdu_index)
2845         target_record_id = "vnfrs:{}:vdur.{}".format(vnf_id, vdu_id)
2846         deployment_info = {
2847             "action_id": action_id,
2848             "nsr_id": nsr_id,
2849             "task_index": task_index,
2850         }
2851
2852         task = Ns._create_task(
2853             deployment_info=deployment_info,
2854             target_id=target_vim,
2855             item="update",
2856             action="EXEC",
2857             target_record=target_record,
2858             target_record_id=target_record_id,
2859             extra_dict=extra_dict,
2860         )
2861         return task
2862
2863     def rebuild_start_stop(
2864         self, session, action_dict, version, nsr_id, *args, **kwargs
2865     ):
2866         task_index = 0
2867         extra_dict = {}
2868         now = time()
2869         action_id = action_dict.get("action_id", str(uuid4()))
2870         step = ""
2871         logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id)
2872         self.logger.debug(logging_text + "Enter")
2873
2874         action = list(action_dict.keys())[0]
2875         task_dict = action_dict.get(action)
2876         vim_vm_id = action_dict.get(action).get("vim_vm_id")
2877
2878         if action_dict.get("stop"):
2879             action = "shutoff"
2880         db_new_tasks = []
2881         try:
2882             step = "lock the operation & do task creation"
2883             with self.write_lock:
2884                 extra_dict["params"] = {
2885                     "vim_vm_id": vim_vm_id,
2886                     "action": action,
2887                 }
2888                 task = self.rebuild_start_stop_task(
2889                     task_dict["vdu_id"],
2890                     task_dict["vnf_id"],
2891                     task_dict["vdu_index"],
2892                     action_id,
2893                     nsr_id,
2894                     task_index,
2895                     task_dict["target_vim"],
2896                     extra_dict,
2897                 )
2898                 db_new_tasks.append(task)
2899                 step = "upload Task to db"
2900                 self.upload_all_tasks(
2901                     db_new_tasks=db_new_tasks,
2902                     now=now,
2903                 )
2904                 self.logger.debug(
2905                     logging_text + "Exit. Created {} tasks".format(len(db_new_tasks))
2906                 )
2907                 return (
2908                     {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
2909                     action_id,
2910                     True,
2911                 )
2912         except Exception as e:
2913             if isinstance(e, (DbException, NsException)):
2914                 self.logger.error(
2915                     logging_text + "Exit Exception while '{}': {}".format(step, e)
2916                 )
2917             else:
2918                 e = traceback_format_exc()
2919                 self.logger.critical(
2920                     logging_text + "Exit Exception while '{}': {}".format(step, e),
2921                     exc_info=True,
2922                 )
2923             raise NsException(e)
2924
2925     def get_deploy(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
2926         nsrs = self.db.get_list("nsrs", {})
2927         return_data = []
2928
2929         for ns in nsrs:
2930             return_data.append({"_id": ns["_id"], "name": ns["name"]})
2931
2932         return return_data, None, True
2933
2934     def get_actions(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
2935         ro_tasks = self.db.get_list("ro_tasks", {"tasks.nsr_id": nsr_id})
2936         return_data = []
2937
2938         for ro_task in ro_tasks:
2939             for task in ro_task["tasks"]:
2940                 if task["action_id"] not in return_data:
2941                     return_data.append(task["action_id"])
2942
2943         return return_data, None, True
2944
2945     def migrate_task(
2946         self, vdu, vnf, vdu_index, action_id, nsr_id, task_index, extra_dict
2947     ):
2948         target_vim, vim_info = next(k_v for k_v in vdu["vim_info"].items())
2949         self._assign_vim(target_vim)
2950         target_record = "vnfrs:{}:vdur.{}".format(vnf["_id"], vdu_index)
2951         target_record_id = "vnfrs:{}:vdur.{}".format(vnf["_id"], vdu["id"])
2952         deployment_info = {
2953             "action_id": action_id,
2954             "nsr_id": nsr_id,
2955             "task_index": task_index,
2956         }
2957
2958         task = Ns._create_task(
2959             deployment_info=deployment_info,
2960             target_id=target_vim,
2961             item="migrate",
2962             action="EXEC",
2963             target_record=target_record,
2964             target_record_id=target_record_id,
2965             extra_dict=extra_dict,
2966         )
2967
2968         return task
2969
2970     def migrate(self, session, indata, version, nsr_id, *args, **kwargs):
2971         task_index = 0
2972         extra_dict = {}
2973         now = time()
2974         action_id = indata.get("action_id", str(uuid4()))
2975         step = ""
2976         logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id)
2977         self.logger.debug(logging_text + "Enter")
2978         try:
2979             vnf_instance_id = indata["vnfInstanceId"]
2980             step = "Getting vnfrs from db"
2981             db_vnfr = self.db.get_one("vnfrs", {"_id": vnf_instance_id})
2982             vdu = indata.get("vdu")
2983             migrateToHost = indata.get("migrateToHost")
2984             db_new_tasks = []
2985
2986             with self.write_lock:
2987                 if vdu is not None:
2988                     vdu_id = indata["vdu"]["vduId"]
2989                     vdu_count_index = indata["vdu"].get("vduCountIndex", 0)
2990                     for vdu_index, vdu in enumerate(db_vnfr["vdur"]):
2991                         if (
2992                             vdu["vdu-id-ref"] == vdu_id
2993                             and vdu["count-index"] == vdu_count_index
2994                         ):
2995                             extra_dict["params"] = {
2996                                 "vim_vm_id": vdu["vim-id"],
2997                                 "migrate_host": migrateToHost,
2998                                 "vdu_vim_info": vdu["vim_info"],
2999                             }
3000                             step = "Creating migration task for vdu:{}".format(vdu)
3001                             task = self.migrate_task(
3002                                 vdu,
3003                                 db_vnfr,
3004                                 vdu_index,
3005                                 action_id,
3006                                 nsr_id,
3007                                 task_index,
3008                                 extra_dict,
3009                             )
3010                             db_new_tasks.append(task)
3011                             task_index += 1
3012                             break
3013                 else:
3014                     for vdu_index, vdu in enumerate(db_vnfr["vdur"]):
3015                         extra_dict["params"] = {
3016                             "vim_vm_id": vdu["vim-id"],
3017                             "migrate_host": migrateToHost,
3018                             "vdu_vim_info": vdu["vim_info"],
3019                         }
3020                         step = "Creating migration task for vdu:{}".format(vdu)
3021                         task = self.migrate_task(
3022                             vdu,
3023                             db_vnfr,
3024                             vdu_index,
3025                             action_id,
3026                             nsr_id,
3027                             task_index,
3028                             extra_dict,
3029                         )
3030                         db_new_tasks.append(task)
3031                         task_index += 1
3032
3033                 self.upload_all_tasks(
3034                     db_new_tasks=db_new_tasks,
3035                     now=now,
3036                 )
3037
3038             self.logger.debug(
3039                 logging_text + "Exit. Created {} tasks".format(len(db_new_tasks))
3040             )
3041             return (
3042                 {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
3043                 action_id,
3044                 True,
3045             )
3046         except Exception as e:
3047             if isinstance(e, (DbException, NsException)):
3048                 self.logger.error(
3049                     logging_text + "Exit Exception while '{}': {}".format(step, e)
3050                 )
3051             else:
3052                 e = traceback_format_exc()
3053                 self.logger.critical(
3054                     logging_text + "Exit Exception while '{}': {}".format(step, e),
3055                     exc_info=True,
3056                 )
3057             raise NsException(e)
3058
3059     def verticalscale_task(
3060         self, vdu, vnf, vdu_index, action_id, nsr_id, task_index, extra_dict
3061     ):
3062         target_vim, vim_info = next(k_v for k_v in vdu["vim_info"].items())
3063         self._assign_vim(target_vim)
3064         target_record = "vnfrs:{}:vdur.{}".format(vnf["_id"], vdu_index)
3065         target_record_id = "vnfrs:{}:vdur.{}".format(vnf["_id"], vdu["id"])
3066         deployment_info = {
3067             "action_id": action_id,
3068             "nsr_id": nsr_id,
3069             "task_index": task_index,
3070         }
3071
3072         task = Ns._create_task(
3073             deployment_info=deployment_info,
3074             target_id=target_vim,
3075             item="verticalscale",
3076             action="EXEC",
3077             target_record=target_record,
3078             target_record_id=target_record_id,
3079             extra_dict=extra_dict,
3080         )
3081         return task
3082
3083     def verticalscale(self, session, indata, version, nsr_id, *args, **kwargs):
3084         task_index = 0
3085         extra_dict = {}
3086         now = time()
3087         action_id = indata.get("action_id", str(uuid4()))
3088         step = ""
3089         logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id)
3090         self.logger.debug(logging_text + "Enter")
3091         try:
3092             VnfFlavorData = indata.get("changeVnfFlavorData")
3093             vnf_instance_id = VnfFlavorData["vnfInstanceId"]
3094             step = "Getting vnfrs from db"
3095             db_vnfr = self.db.get_one("vnfrs", {"_id": vnf_instance_id})
3096             vduid = VnfFlavorData["additionalParams"]["vduid"]
3097             vduCountIndex = VnfFlavorData["additionalParams"]["vduCountIndex"]
3098             virtualMemory = VnfFlavorData["additionalParams"]["virtualMemory"]
3099             numVirtualCpu = VnfFlavorData["additionalParams"]["numVirtualCpu"]
3100             sizeOfStorage = VnfFlavorData["additionalParams"]["sizeOfStorage"]
3101             flavor_dict = {
3102                 "name": vduid + "-flv",
3103                 "ram": virtualMemory,
3104                 "vcpus": numVirtualCpu,
3105                 "disk": sizeOfStorage,
3106             }
3107             db_new_tasks = []
3108             step = "Creating Tasks for vertical scaling"
3109             with self.write_lock:
3110                 for vdu_index, vdu in enumerate(db_vnfr["vdur"]):
3111                     if (
3112                         vdu["vdu-id-ref"] == vduid
3113                         and vdu["count-index"] == vduCountIndex
3114                     ):
3115                         extra_dict["params"] = {
3116                             "vim_vm_id": vdu["vim-id"],
3117                             "flavor_dict": flavor_dict,
3118                         }
3119                         task = self.verticalscale_task(
3120                             vdu,
3121                             db_vnfr,
3122                             vdu_index,
3123                             action_id,
3124                             nsr_id,
3125                             task_index,
3126                             extra_dict,
3127                         )
3128                         db_new_tasks.append(task)
3129                         task_index += 1
3130                         break
3131                 self.upload_all_tasks(
3132                     db_new_tasks=db_new_tasks,
3133                     now=now,
3134                 )
3135             self.logger.debug(
3136                 logging_text + "Exit. Created {} tasks".format(len(db_new_tasks))
3137             )
3138             return (
3139                 {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
3140                 action_id,
3141                 True,
3142             )
3143         except Exception as e:
3144             if isinstance(e, (DbException, NsException)):
3145                 self.logger.error(
3146                     logging_text + "Exit Exception while '{}': {}".format(step, e)
3147                 )
3148             else:
3149                 e = traceback_format_exc()
3150                 self.logger.critical(
3151                     logging_text + "Exit Exception while '{}': {}".format(step, e),
3152                     exc_info=True,
3153                 )
3154             raise NsException(e)