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