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"][:16],
1841                 vnfr["member-vnf-index-ref"][:16],
1842                 target_vdu["vdu-name"][:32],
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                 net_item["ip_address"] = interface["ip-address"]
2005
2006             if interface.get("mac-address"):
2007                 net_item["mac_address"] = interface["mac-address"]
2008
2009             net_list.append(net_item)
2010
2011             if interface.get("mgmt-vnf"):
2012                 extra_dict["mgmt_vnf_interface"] = iface_index
2013             elif interface.get("mgmt-interface"):
2014                 extra_dict["mgmt_vdu_interface"] = iface_index
2015
2016         # cloud config
2017         cloud_config = {}
2018
2019         if existing_vdu.get("cloud-init"):
2020             if existing_vdu["cloud-init"] not in vdu2cloud_init:
2021                 vdu2cloud_init[existing_vdu["cloud-init"]] = Ns._get_cloud_init(
2022                     db=db,
2023                     fs=fs,
2024                     location=existing_vdu["cloud-init"],
2025                 )
2026
2027             cloud_content_ = vdu2cloud_init[existing_vdu["cloud-init"]]
2028             cloud_config["user-data"] = Ns._parse_jinja2(
2029                 cloud_init_content=cloud_content_,
2030                 params=existing_vdu.get("additionalParams"),
2031                 context=existing_vdu["cloud-init"],
2032             )
2033
2034         if existing_vdu.get("boot-data-drive"):
2035             cloud_config["boot-data-drive"] = existing_vdu.get("boot-data-drive")
2036
2037         ssh_keys = []
2038
2039         if existing_vdu.get("ssh-keys"):
2040             ssh_keys += existing_vdu.get("ssh-keys")
2041
2042         if existing_vdu.get("ssh-access-required"):
2043             ssh_keys.append(ro_nsr_public_key)
2044
2045         if ssh_keys:
2046             cloud_config["key-pairs"] = ssh_keys
2047
2048         disk_list = []
2049         for vol_id in vim_details.get("os-extended-volumes:volumes_attached", []):
2050             disk_list.append({"vim_id": vol_id["id"]})
2051
2052         affinity_group_list = []
2053
2054         if existing_vdu.get("affinity-or-anti-affinity-group-id"):
2055             affinity_group = {}
2056             for affinity_group_id in existing_vdu["affinity-or-anti-affinity-group-id"]:
2057                 for group in db_nsr.get("affinity-or-anti-affinity-group"):
2058                     if (
2059                         group["id"] == affinity_group_id
2060                         and group["vim_info"][target_id].get("vim_id", None) is not None
2061                     ):
2062                         affinity_group["affinity_group_id"] = group["vim_info"][
2063                             target_id
2064                         ].get("vim_id", None)
2065                         affinity_group_list.append(affinity_group)
2066
2067         extra_dict["params"] = {
2068             "name": "{}-{}-{}-{}".format(
2069                 db_nsr["name"][:16],
2070                 vnfr["member-vnf-index-ref"][:16],
2071                 existing_vdu["vdu-name"][:32],
2072                 existing_vdu.get("count-index") or 0,
2073             ),
2074             "description": existing_vdu["vdu-name"],
2075             "start": True,
2076             "image_id": vim_details["image"]["id"],
2077             "flavor_id": vim_details["flavor"]["id"],
2078             "affinity_group_list": affinity_group_list,
2079             "net_list": net_list,
2080             "cloud_config": cloud_config or None,
2081             "disk_list": disk_list,
2082             "availability_zone_index": None,  # TODO
2083             "availability_zone_list": None,  # TODO
2084         }
2085
2086         return extra_dict
2087
2088     def calculate_diff_items(
2089         self,
2090         indata,
2091         db_nsr,
2092         db_ro_nsr,
2093         db_nsr_update,
2094         item,
2095         tasks_by_target_record_id,
2096         action_id,
2097         nsr_id,
2098         task_index,
2099         vnfr_id=None,
2100         vnfr=None,
2101     ):
2102         """Function that returns the incremental changes (creation, deletion)
2103         related to a specific item `item` to be done. This function should be
2104         called for NS instantiation, NS termination, NS update to add a new VNF
2105         or a new VLD, remove a VNF or VLD, etc.
2106         Item can be `net`, `flavor`, `image` or `vdu`.
2107         It takes a list of target items from indata (which came from the REST API)
2108         and compares with the existing items from db_ro_nsr, identifying the
2109         incremental changes to be done. During the comparison, it calls the method
2110         `process_params` (which was passed as parameter, and is particular for each
2111         `item`)
2112
2113         Args:
2114             indata (Dict[str, Any]): deployment info
2115             db_nsr: NSR record from DB
2116             db_ro_nsr (Dict[str, Any]): record from "ro_nsrs"
2117             db_nsr_update (Dict[str, Any]): NSR info to update in DB
2118             item (str): element to process (net, vdu...)
2119             tasks_by_target_record_id (Dict[str, Any]):
2120                 [<target_record_id>, <task>]
2121             action_id (str): action id
2122             nsr_id (str): NSR id
2123             task_index (number): task index to add to task name
2124             vnfr_id (str): VNFR id
2125             vnfr (Dict[str, Any]): VNFR info
2126
2127         Returns:
2128             List: list with the incremental changes (deletes, creates) for each item
2129             number: current task index
2130         """
2131
2132         diff_items = []
2133         db_path = ""
2134         db_record = ""
2135         target_list = []
2136         existing_list = []
2137         process_params = None
2138         vdu2cloud_init = indata.get("cloud_init_content") or {}
2139         ro_nsr_public_key = db_ro_nsr["public_key"]
2140         # According to the type of item, the path, the target_list,
2141         # the existing_list and the method to process params are set
2142         db_path = self.db_path_map[item]
2143         process_params = self.process_params_function_map[item]
2144
2145         if item in ("sfp", "classification", "sf", "sfi"):
2146             db_record = "nsrs:{}:{}".format(nsr_id, db_path)
2147             target_vnffg = indata.get("vnffg", [])[0]
2148             target_list = target_vnffg[item]
2149             existing_list = db_nsr.get(item, [])
2150         elif item in ("net", "vdu"):
2151             # This case is specific for the NS VLD (not applied to VDU)
2152             if vnfr is None:
2153                 db_record = "nsrs:{}:{}".format(nsr_id, db_path)
2154                 target_list = indata.get("ns", []).get(db_path, [])
2155                 existing_list = db_nsr.get(db_path, [])
2156             # This case is common for VNF VLDs and VNF VDUs
2157             else:
2158                 db_record = "vnfrs:{}:{}".format(vnfr_id, db_path)
2159                 target_vnf = next(
2160                     (vnf for vnf in indata.get("vnf", ()) if vnf["_id"] == vnfr_id),
2161                     None,
2162                 )
2163                 target_list = target_vnf.get(db_path, []) if target_vnf else []
2164                 existing_list = vnfr.get(db_path, [])
2165         elif item in (
2166             "image",
2167             "flavor",
2168             "affinity-or-anti-affinity-group",
2169             "shared-volumes",
2170         ):
2171             db_record = "nsrs:{}:{}".format(nsr_id, db_path)
2172             target_list = indata.get(item, [])
2173             existing_list = db_nsr.get(item, [])
2174         else:
2175             raise NsException("Item not supported: {}", item)
2176         # ensure all the target_list elements has an "id". If not assign the index as id
2177         if target_list is None:
2178             target_list = []
2179         for target_index, tl in enumerate(target_list):
2180             if tl and not tl.get("id"):
2181                 tl["id"] = str(target_index)
2182         # step 1 items (networks,vdus,...) to be deleted/updated
2183         for item_index, existing_item in enumerate(existing_list):
2184             target_item = next(
2185                 (t for t in target_list if t["id"] == existing_item["id"]),
2186                 None,
2187             )
2188             for target_vim, existing_viminfo in existing_item.get(
2189                 "vim_info", {}
2190             ).items():
2191                 if existing_viminfo is None:
2192                     continue
2193
2194                 if target_item:
2195                     target_viminfo = target_item.get("vim_info", {}).get(target_vim)
2196                 else:
2197                     target_viminfo = None
2198
2199                 if target_viminfo is None:
2200                     # must be deleted
2201                     self._assign_vim(target_vim)
2202                     target_record_id = "{}.{}".format(db_record, existing_item["id"])
2203                     item_ = item
2204
2205                     if target_vim.startswith("sdn") or target_vim.startswith("wim"):
2206                         # item must be sdn-net instead of net if target_vim is a sdn
2207                         item_ = "sdn_net"
2208                         target_record_id += ".sdn"
2209
2210                     deployment_info = {
2211                         "action_id": action_id,
2212                         "nsr_id": nsr_id,
2213                         "task_index": task_index,
2214                     }
2215
2216                     diff_items.append(
2217                         {
2218                             "deployment_info": deployment_info,
2219                             "target_id": target_vim,
2220                             "item": item_,
2221                             "action": "DELETE",
2222                             "target_record": f"{db_record}.{item_index}.vim_info.{target_vim}",
2223                             "target_record_id": target_record_id,
2224                         }
2225                     )
2226                     task_index += 1
2227
2228         # step 2 items (networks,vdus,...) to be created
2229         for target_item in target_list:
2230             item_index = -1
2231             for item_index, existing_item in enumerate(existing_list):
2232                 if existing_item["id"] == target_item["id"]:
2233                     break
2234             else:
2235                 item_index += 1
2236                 db_nsr_update[db_path + ".{}".format(item_index)] = target_item
2237                 existing_list.append(target_item)
2238                 existing_item = None
2239
2240             for target_vim, target_viminfo in target_item.get("vim_info", {}).items():
2241                 existing_viminfo = None
2242
2243                 if existing_item:
2244                     existing_viminfo = existing_item.get("vim_info", {}).get(target_vim)
2245
2246                 if existing_viminfo is not None:
2247                     continue
2248
2249                 target_record_id = "{}.{}".format(db_record, target_item["id"])
2250                 item_ = item
2251
2252                 if target_vim.startswith("sdn") or target_vim.startswith("wim"):
2253                     # item must be sdn-net instead of net if target_vim is a sdn
2254                     item_ = "sdn_net"
2255                     target_record_id += ".sdn"
2256
2257                 kwargs = {}
2258                 self.logger.debug(
2259                     "ns.calculate_diff_items target_item={}".format(target_item)
2260                 )
2261                 if process_params == Ns._process_flavor_params:
2262                     kwargs.update(
2263                         {
2264                             "db": self.db,
2265                         }
2266                     )
2267                     self.logger.debug(
2268                         "calculate_diff_items for flavor kwargs={}".format(kwargs)
2269                     )
2270
2271                 if process_params == Ns._process_vdu_params:
2272                     self.logger.debug("calculate_diff_items self.fs={}".format(self.fs))
2273                     kwargs.update(
2274                         {
2275                             "vnfr_id": vnfr_id,
2276                             "nsr_id": nsr_id,
2277                             "vnfr": vnfr,
2278                             "vdu2cloud_init": vdu2cloud_init,
2279                             "tasks_by_target_record_id": tasks_by_target_record_id,
2280                             "logger": self.logger,
2281                             "db": self.db,
2282                             "fs": self.fs,
2283                             "ro_nsr_public_key": ro_nsr_public_key,
2284                         }
2285                     )
2286                     self.logger.debug("calculate_diff_items kwargs={}".format(kwargs))
2287                 if (
2288                     process_params == Ns._process_sfi_params
2289                     or Ns._process_sf_params
2290                     or Ns._process_classification_params
2291                     or Ns._process_sfp_params
2292                 ):
2293                     kwargs.update({"nsr_id": nsr_id, "db": self.db})
2294
2295                     self.logger.debug("calculate_diff_items kwargs={}".format(kwargs))
2296
2297                 extra_dict = process_params(
2298                     target_item,
2299                     indata,
2300                     target_viminfo,
2301                     target_record_id,
2302                     **kwargs,
2303                 )
2304                 self._assign_vim(target_vim)
2305
2306                 deployment_info = {
2307                     "action_id": action_id,
2308                     "nsr_id": nsr_id,
2309                     "task_index": task_index,
2310                 }
2311
2312                 new_item = {
2313                     "deployment_info": deployment_info,
2314                     "target_id": target_vim,
2315                     "item": item_,
2316                     "action": "CREATE",
2317                     "target_record": f"{db_record}.{item_index}.vim_info.{target_vim}",
2318                     "target_record_id": target_record_id,
2319                     "extra_dict": extra_dict,
2320                     "common_id": target_item.get("common_id", None),
2321                 }
2322                 diff_items.append(new_item)
2323                 tasks_by_target_record_id[target_record_id] = new_item
2324                 task_index += 1
2325
2326                 db_nsr_update[db_path + ".{}".format(item_index)] = target_item
2327
2328         return diff_items, task_index
2329
2330     def _process_vnfgd_sfp(self, sfp):
2331         processed_sfp = {}
2332         # getting sfp name, sfs and classifications in sfp to store it in processed_sfp
2333         processed_sfp["id"] = sfp["id"]
2334         sfs_in_sfp = [
2335             sf["id"] for sf in sfp.get("position-desc-id", [])[0].get("cp-profile-id")
2336         ]
2337         classifications_in_sfp = [
2338             classi["id"]
2339             for classi in sfp.get("position-desc-id", [])[0].get("match-attributes")
2340         ]
2341
2342         # creating a list of sfp with sfs and classifications
2343         processed_sfp["sfs"] = sfs_in_sfp
2344         processed_sfp["classifications"] = classifications_in_sfp
2345
2346         return processed_sfp
2347
2348     def _process_vnfgd_sf(self, sf):
2349         processed_sf = {}
2350         # getting name of sf
2351         processed_sf["id"] = sf["id"]
2352         # getting sfis in sf
2353         sfis_in_sf = sf.get("constituent-profile-elements")
2354         sorted_sfis = sorted(sfis_in_sf, key=lambda i: i["order"])
2355         # getting sfis names
2356         processed_sf["sfis"] = [sfi["id"] for sfi in sorted_sfis]
2357
2358         return processed_sf
2359
2360     def _process_vnfgd_sfi(self, sfi, db_vnfrs):
2361         processed_sfi = {}
2362         # getting name of sfi
2363         processed_sfi["id"] = sfi["id"]
2364
2365         # getting ports in sfi
2366         ingress_port = sfi["ingress-constituent-cpd-id"]
2367         egress_port = sfi["egress-constituent-cpd-id"]
2368         sfi_vnf_member_index = sfi["constituent-base-element-id"]
2369
2370         processed_sfi["ingress_port"] = ingress_port
2371         processed_sfi["egress_port"] = egress_port
2372
2373         all_vnfrs = db_vnfrs.values()
2374
2375         sfi_vnfr = [
2376             element
2377             for element in all_vnfrs
2378             if element["member-vnf-index-ref"] == sfi_vnf_member_index
2379         ]
2380         processed_sfi["vnfr_id"] = sfi_vnfr[0]["id"]
2381
2382         sfi_vnfr_cp = sfi_vnfr[0]["connection-point"]
2383
2384         ingress_port_index = [
2385             c for c, element in enumerate(sfi_vnfr_cp) if element["id"] == ingress_port
2386         ]
2387         ingress_port_index = ingress_port_index[0]
2388
2389         processed_sfi["vdur_id"] = sfi_vnfr_cp[ingress_port_index][
2390             "connection-point-vdu-id"
2391         ]
2392         processed_sfi["ingress_port_index"] = ingress_port_index
2393         processed_sfi["egress_port_index"] = ingress_port_index
2394
2395         if egress_port != ingress_port:
2396             egress_port_index = [
2397                 c
2398                 for c, element in enumerate(sfi_vnfr_cp)
2399                 if element["id"] == egress_port
2400             ]
2401             processed_sfi["egress_port_index"] = egress_port_index
2402
2403         return processed_sfi
2404
2405     def _process_vnfgd_classification(self, classification, db_vnfrs):
2406         processed_classification = {}
2407
2408         processed_classification = deepcopy(classification)
2409         classi_vnf_member_index = processed_classification[
2410             "constituent-base-element-id"
2411         ]
2412         logical_source_port = processed_classification["constituent-cpd-id"]
2413
2414         all_vnfrs = db_vnfrs.values()
2415
2416         classi_vnfr = [
2417             element
2418             for element in all_vnfrs
2419             if element["member-vnf-index-ref"] == classi_vnf_member_index
2420         ]
2421         processed_classification["vnfr_id"] = classi_vnfr[0]["id"]
2422
2423         classi_vnfr_cp = classi_vnfr[0]["connection-point"]
2424
2425         ingress_port_index = [
2426             c
2427             for c, element in enumerate(classi_vnfr_cp)
2428             if element["id"] == logical_source_port
2429         ]
2430         ingress_port_index = ingress_port_index[0]
2431
2432         processed_classification["ingress_port_index"] = ingress_port_index
2433         processed_classification["vdur_id"] = classi_vnfr_cp[ingress_port_index][
2434             "connection-point-vdu-id"
2435         ]
2436
2437         return processed_classification
2438
2439     def _update_db_nsr_with_vnffg(self, processed_vnffg, vim_info, nsr_id):
2440         """This method used to add viminfo dict to sfi, sf sfp and classification in indata and count info in db_nsr.
2441
2442         Args:
2443             processed_vnffg (Dict[str, Any]): deployment info
2444             vim_info (Dict): dictionary to store VIM resource information
2445             nsr_id (str): NSR id
2446
2447         Returns: None
2448         """
2449
2450         nsr_sfi = {}
2451         nsr_sf = {}
2452         nsr_sfp = {}
2453         nsr_classification = {}
2454         db_nsr_vnffg = deepcopy(processed_vnffg)
2455
2456         for count, sfi in enumerate(processed_vnffg["sfi"]):
2457             sfi["vim_info"] = vim_info
2458             sfi_count = "sfi.{}".format(count)
2459             nsr_sfi[sfi_count] = db_nsr_vnffg["sfi"][count]
2460
2461         self.db.set_list("nsrs", {"_id": nsr_id}, nsr_sfi)
2462
2463         for count, sf in enumerate(processed_vnffg["sf"]):
2464             sf["vim_info"] = vim_info
2465             sf_count = "sf.{}".format(count)
2466             nsr_sf[sf_count] = db_nsr_vnffg["sf"][count]
2467
2468         self.db.set_list("nsrs", {"_id": nsr_id}, nsr_sf)
2469
2470         for count, sfp in enumerate(processed_vnffg["sfp"]):
2471             sfp["vim_info"] = vim_info
2472             sfp_count = "sfp.{}".format(count)
2473             nsr_sfp[sfp_count] = db_nsr_vnffg["sfp"][count]
2474
2475         self.db.set_list("nsrs", {"_id": nsr_id}, nsr_sfp)
2476
2477         for count, classi in enumerate(processed_vnffg["classification"]):
2478             classi["vim_info"] = vim_info
2479             classification_count = "classification.{}".format(count)
2480             nsr_classification[classification_count] = db_nsr_vnffg["classification"][
2481                 count
2482             ]
2483
2484             self.db.set_list("nsrs", {"_id": nsr_id}, nsr_classification)
2485
2486     def process_vnffgd_descriptor(
2487         self,
2488         indata: dict,
2489         nsr_id: str,
2490         db_nsr: dict,
2491         db_vnfrs: dict,
2492     ) -> dict:
2493         """This method used to process vnffgd parameters from descriptor.
2494
2495         Args:
2496             indata (Dict[str, Any]): deployment info
2497             nsr_id (str): NSR id
2498             db_nsr: NSR record from DB
2499             db_vnfrs: VNFRS record from DB
2500
2501         Returns:
2502             Dict: Processed vnffg parameters.
2503         """
2504
2505         processed_vnffg = {}
2506         vnffgd = db_nsr.get("nsd", {}).get("vnffgd")
2507         vnf_list = indata.get("vnf", [])
2508         vim_text = ""
2509
2510         if vnf_list:
2511             vim_text = "vim:" + vnf_list[0].get("vim-account-id", "")
2512
2513         vim_info = {}
2514         vim_info[vim_text] = {}
2515         processed_sfps = []
2516         processed_classifications = []
2517         processed_sfs = []
2518         processed_sfis = []
2519
2520         # setting up intial empty entries for vnffg items in mongodb.
2521         self.db.set_list(
2522             "nsrs",
2523             {"_id": nsr_id},
2524             {
2525                 "sfi": [],
2526                 "sf": [],
2527                 "sfp": [],
2528                 "classification": [],
2529             },
2530         )
2531
2532         vnffg = vnffgd[0]
2533         # getting sfps
2534         sfps = vnffg.get("nfpd")
2535         for sfp in sfps:
2536             processed_sfp = self._process_vnfgd_sfp(sfp)
2537             # appending the list of processed sfps
2538             processed_sfps.append(processed_sfp)
2539
2540             # getting sfs in sfp
2541             sfs = sfp.get("position-desc-id")[0].get("cp-profile-id")
2542             for sf in sfs:
2543                 processed_sf = self._process_vnfgd_sf(sf)
2544
2545                 # appending the list of processed sfs
2546                 processed_sfs.append(processed_sf)
2547
2548                 # getting sfis in sf
2549                 sfis_in_sf = sf.get("constituent-profile-elements")
2550                 sorted_sfis = sorted(sfis_in_sf, key=lambda i: i["order"])
2551
2552                 for sfi in sorted_sfis:
2553                     processed_sfi = self._process_vnfgd_sfi(sfi, db_vnfrs)
2554
2555                     processed_sfis.append(processed_sfi)
2556
2557             classifications = sfp.get("position-desc-id")[0].get("match-attributes")
2558             # getting classifications from sfp
2559             for classification in classifications:
2560                 processed_classification = self._process_vnfgd_classification(
2561                     classification, db_vnfrs
2562                 )
2563
2564                 processed_classifications.append(processed_classification)
2565
2566         processed_vnffg["sfi"] = processed_sfis
2567         processed_vnffg["sf"] = processed_sfs
2568         processed_vnffg["classification"] = processed_classifications
2569         processed_vnffg["sfp"] = processed_sfps
2570
2571         # adding viminfo dict to sfi, sf sfp and classification
2572         self._update_db_nsr_with_vnffg(processed_vnffg, vim_info, nsr_id)
2573
2574         # updating indata with vnffg porcessed parameters
2575         indata["vnffg"].append(processed_vnffg)
2576
2577     def calculate_all_differences_to_deploy(
2578         self,
2579         indata,
2580         nsr_id,
2581         db_nsr,
2582         db_vnfrs,
2583         db_ro_nsr,
2584         db_nsr_update,
2585         db_vnfrs_update,
2586         action_id,
2587         tasks_by_target_record_id,
2588     ):
2589         """This method calculates the ordered list of items (`changes_list`)
2590         to be created and deleted.
2591
2592         Args:
2593             indata (Dict[str, Any]): deployment info
2594             nsr_id (str): NSR id
2595             db_nsr: NSR record from DB
2596             db_vnfrs: VNFRS record from DB
2597             db_ro_nsr (Dict[str, Any]): record from "ro_nsrs"
2598             db_nsr_update (Dict[str, Any]): NSR info to update in DB
2599             db_vnfrs_update (Dict[str, Any]): VNFRS info to update in DB
2600             action_id (str): action id
2601             tasks_by_target_record_id (Dict[str, Any]):
2602                 [<target_record_id>, <task>]
2603
2604         Returns:
2605             List: ordered list of items to be created and deleted.
2606         """
2607
2608         task_index = 0
2609         # set list with diffs:
2610         changes_list = []
2611
2612         # processing vnffg from descriptor parameter
2613         vnffgd = db_nsr.get("nsd").get("vnffgd")
2614         if vnffgd is not None:
2615             indata["vnffg"] = []
2616             vnf_list = indata["vnf"]
2617             processed_vnffg = {}
2618
2619             # in case of ns-delete
2620             if not vnf_list:
2621                 processed_vnffg["sfi"] = []
2622                 processed_vnffg["sf"] = []
2623                 processed_vnffg["classification"] = []
2624                 processed_vnffg["sfp"] = []
2625
2626                 indata["vnffg"].append(processed_vnffg)
2627
2628             else:
2629                 self.process_vnffgd_descriptor(
2630                     indata=indata,
2631                     nsr_id=nsr_id,
2632                     db_nsr=db_nsr,
2633                     db_vnfrs=db_vnfrs,
2634                 )
2635
2636                 # getting updated db_nsr having vnffg parameters
2637                 db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
2638
2639                 self.logger.debug(
2640                     "After processing vnffd parameters indata={} nsr={}".format(
2641                         indata, db_nsr
2642                     )
2643                 )
2644
2645             for item in ["sfp", "classification", "sf", "sfi"]:
2646                 self.logger.debug("process NS={} {}".format(nsr_id, item))
2647                 diff_items, task_index = self.calculate_diff_items(
2648                     indata=indata,
2649                     db_nsr=db_nsr,
2650                     db_ro_nsr=db_ro_nsr,
2651                     db_nsr_update=db_nsr_update,
2652                     item=item,
2653                     tasks_by_target_record_id=tasks_by_target_record_id,
2654                     action_id=action_id,
2655                     nsr_id=nsr_id,
2656                     task_index=task_index,
2657                     vnfr_id=None,
2658                 )
2659                 changes_list += diff_items
2660
2661         # NS vld, image and flavor
2662         for item in [
2663             "net",
2664             "image",
2665             "flavor",
2666             "affinity-or-anti-affinity-group",
2667         ]:
2668             self.logger.debug("process NS={} {}".format(nsr_id, item))
2669             diff_items, task_index = self.calculate_diff_items(
2670                 indata=indata,
2671                 db_nsr=db_nsr,
2672                 db_ro_nsr=db_ro_nsr,
2673                 db_nsr_update=db_nsr_update,
2674                 item=item,
2675                 tasks_by_target_record_id=tasks_by_target_record_id,
2676                 action_id=action_id,
2677                 nsr_id=nsr_id,
2678                 task_index=task_index,
2679                 vnfr_id=None,
2680             )
2681             changes_list += diff_items
2682
2683         # VNF vlds and vdus
2684         for vnfr_id, vnfr in db_vnfrs.items():
2685             # vnfr_id need to be set as global variable for among others nested method _process_vdu_params
2686             for item in ["net", "vdu", "shared-volumes"]:
2687                 self.logger.debug("process VNF={} {}".format(vnfr_id, item))
2688                 diff_items, task_index = self.calculate_diff_items(
2689                     indata=indata,
2690                     db_nsr=db_nsr,
2691                     db_ro_nsr=db_ro_nsr,
2692                     db_nsr_update=db_vnfrs_update[vnfr["_id"]],
2693                     item=item,
2694                     tasks_by_target_record_id=tasks_by_target_record_id,
2695                     action_id=action_id,
2696                     nsr_id=nsr_id,
2697                     task_index=task_index,
2698                     vnfr_id=vnfr_id,
2699                     vnfr=vnfr,
2700                 )
2701                 changes_list += diff_items
2702
2703         return changes_list
2704
2705     def define_all_tasks(
2706         self,
2707         changes_list,
2708         db_new_tasks,
2709         tasks_by_target_record_id,
2710     ):
2711         """Function to create all the task structures obtanied from
2712         the method calculate_all_differences_to_deploy
2713
2714         Args:
2715             changes_list (List): ordered list of items to be created or deleted
2716             db_new_tasks (List): tasks list to be created
2717             action_id (str): action id
2718             tasks_by_target_record_id (Dict[str, Any]):
2719                 [<target_record_id>, <task>]
2720
2721         """
2722
2723         for change in changes_list:
2724             task = Ns._create_task(
2725                 deployment_info=change["deployment_info"],
2726                 target_id=change["target_id"],
2727                 item=change["item"],
2728                 action=change["action"],
2729                 target_record=change["target_record"],
2730                 target_record_id=change["target_record_id"],
2731                 extra_dict=change.get("extra_dict", None),
2732             )
2733
2734             self.logger.debug("ns.define_all_tasks task={}".format(task))
2735             tasks_by_target_record_id[change["target_record_id"]] = task
2736             db_new_tasks.append(task)
2737
2738             if change.get("common_id"):
2739                 task["common_id"] = change["common_id"]
2740
2741     def upload_all_tasks(
2742         self,
2743         db_new_tasks,
2744         now,
2745     ):
2746         """Function to save all tasks in the common DB
2747
2748         Args:
2749             db_new_tasks (List): tasks list to be created
2750             now (time): current time
2751
2752         """
2753
2754         nb_ro_tasks = 0  # for logging
2755
2756         for db_task in db_new_tasks:
2757             target_id = db_task.pop("target_id")
2758             common_id = db_task.get("common_id")
2759
2760             # Do not chek tasks with vim_status DELETED
2761             # because in manual heealing there are two tasks for the same vdur:
2762             #   one with vim_status deleted and the other one with the actual VM status.
2763
2764             if common_id:
2765                 if self.db.set_one(
2766                     "ro_tasks",
2767                     q_filter={
2768                         "target_id": target_id,
2769                         "tasks.common_id": common_id,
2770                         "vim_info.vim_status.ne": "DELETED",
2771                     },
2772                     update_dict={"to_check_at": now, "modified_at": now},
2773                     push={"tasks": db_task},
2774                     fail_on_empty=False,
2775                 ):
2776                     continue
2777
2778             if not self.db.set_one(
2779                 "ro_tasks",
2780                 q_filter={
2781                     "target_id": target_id,
2782                     "tasks.target_record": db_task["target_record"],
2783                     "vim_info.vim_status.ne": "DELETED",
2784                 },
2785                 update_dict={"to_check_at": now, "modified_at": now},
2786                 push={"tasks": db_task},
2787                 fail_on_empty=False,
2788             ):
2789                 # Create a ro_task
2790                 self.logger.debug("Updating database, Creating ro_tasks")
2791                 db_ro_task = Ns._create_ro_task(target_id, db_task)
2792                 nb_ro_tasks += 1
2793                 self.db.create("ro_tasks", db_ro_task)
2794
2795         self.logger.debug(
2796             "Created {} ro_tasks; {} tasks - db_new_tasks={}".format(
2797                 nb_ro_tasks, len(db_new_tasks), db_new_tasks
2798             )
2799         )
2800
2801     def upload_recreate_tasks(
2802         self,
2803         db_new_tasks,
2804         now,
2805     ):
2806         """Function to save recreate tasks in the common DB
2807
2808         Args:
2809             db_new_tasks (List): tasks list to be created
2810             now (time): current time
2811
2812         """
2813
2814         nb_ro_tasks = 0  # for logging
2815
2816         for db_task in db_new_tasks:
2817             target_id = db_task.pop("target_id")
2818             self.logger.debug("target_id={} db_task={}".format(target_id, db_task))
2819
2820             action = db_task.get("action", None)
2821
2822             # Create a ro_task
2823             self.logger.debug("Updating database, Creating ro_tasks")
2824             db_ro_task = Ns._create_ro_task(target_id, db_task)
2825
2826             # If DELETE task: the associated created items should be removed
2827             # (except persistent volumes):
2828             if action == "DELETE":
2829                 db_ro_task["vim_info"]["created"] = True
2830                 db_ro_task["vim_info"]["created_items"] = db_task.get(
2831                     "created_items", {}
2832                 )
2833                 db_ro_task["vim_info"]["volumes_to_hold"] = db_task.get(
2834                     "volumes_to_hold", []
2835                 )
2836                 db_ro_task["vim_info"]["vim_id"] = db_task.get("vim_id", None)
2837
2838             nb_ro_tasks += 1
2839             self.logger.debug("upload_all_tasks db_ro_task={}".format(db_ro_task))
2840             self.db.create("ro_tasks", db_ro_task)
2841
2842         self.logger.debug(
2843             "Created {} ro_tasks; {} tasks - db_new_tasks={}".format(
2844                 nb_ro_tasks, len(db_new_tasks), db_new_tasks
2845             )
2846         )
2847
2848     def _prepare_created_items_for_healing(
2849         self,
2850         nsr_id,
2851         target_record,
2852     ):
2853         created_items = {}
2854         # Get created_items from ro_task
2855         ro_tasks = self.db.get_list("ro_tasks", {"tasks.nsr_id": nsr_id})
2856         for ro_task in ro_tasks:
2857             for task in ro_task["tasks"]:
2858                 if (
2859                     task["target_record"] == target_record
2860                     and task["action"] == "CREATE"
2861                     and ro_task["vim_info"]["created_items"]
2862                 ):
2863                     created_items = ro_task["vim_info"]["created_items"]
2864                     break
2865
2866         return created_items
2867
2868     def _prepare_persistent_volumes_for_healing(
2869         self,
2870         target_id,
2871         existing_vdu,
2872     ):
2873         # The associated volumes of the VM shouldn't be removed
2874         volumes_list = []
2875         vim_details = {}
2876         vim_details_text = existing_vdu["vim_info"][target_id].get("vim_details", None)
2877         if vim_details_text:
2878             vim_details = yaml.safe_load(f"{vim_details_text}")
2879
2880             for vol_id in vim_details.get("os-extended-volumes:volumes_attached", []):
2881                 volumes_list.append(vol_id["id"])
2882
2883         return volumes_list
2884
2885     def prepare_changes_to_recreate(
2886         self,
2887         indata,
2888         nsr_id,
2889         db_nsr,
2890         db_vnfrs,
2891         db_ro_nsr,
2892         action_id,
2893         tasks_by_target_record_id,
2894     ):
2895         """This method will obtain an ordered list of items (`changes_list`)
2896         to be created and deleted to meet the recreate request.
2897         """
2898
2899         self.logger.debug(
2900             "ns.prepare_changes_to_recreate nsr_id={} indata={}".format(nsr_id, indata)
2901         )
2902
2903         task_index = 0
2904         # set list with diffs:
2905         changes_list = []
2906         db_path = self.db_path_map["vdu"]
2907         target_list = indata.get("healVnfData", {})
2908         vdu2cloud_init = indata.get("cloud_init_content") or {}
2909         ro_nsr_public_key = db_ro_nsr["public_key"]
2910
2911         # Check each VNF of the target
2912         for target_vnf in target_list:
2913             # Find this VNF in the list from DB, raise exception if vnfInstanceId is not found
2914             vnfr_id = target_vnf["vnfInstanceId"]
2915             existing_vnf = db_vnfrs.get(vnfr_id, {})
2916             db_record = "vnfrs:{}:{}".format(vnfr_id, db_path)
2917             # vim_account_id = existing_vnf.get("vim-account-id", "")
2918
2919             target_vdus = target_vnf.get("additionalParams", {}).get("vdu", [])
2920             # Check each VDU of this VNF
2921             if not target_vdus:
2922                 # Create target_vdu_list from DB, if VDUs are not specified
2923                 target_vdus = []
2924                 for existing_vdu in existing_vnf.get("vdur"):
2925                     vdu_name = existing_vdu.get("vdu-name", None)
2926                     vdu_index = existing_vdu.get("count-index", 0)
2927                     vdu_to_be_healed = {"vdu-id": vdu_name, "count-index": vdu_index}
2928                     target_vdus.append(vdu_to_be_healed)
2929             for target_vdu in target_vdus:
2930                 vdu_name = target_vdu.get("vdu-id", None)
2931                 # For multi instance VDU count-index is mandatory
2932                 # For single session VDU count-indes is 0
2933                 count_index = target_vdu.get("count-index", 0)
2934                 item_index = 0
2935                 existing_instance = {}
2936                 if existing_vnf:
2937                     for instance in existing_vnf.get("vdur", {}):
2938                         if (
2939                             instance["vdu-name"] == vdu_name
2940                             and instance["count-index"] == count_index
2941                         ):
2942                             existing_instance = instance
2943                             break
2944                         else:
2945                             item_index += 1
2946
2947                 target_record_id = "{}.{}".format(db_record, existing_instance["id"])
2948
2949                 # The target VIM is the one already existing in DB to recreate
2950                 for target_vim, target_viminfo in existing_instance.get(
2951                     "vim_info", {}
2952                 ).items():
2953                     # step 1 vdu to be deleted
2954                     self._assign_vim(target_vim)
2955                     deployment_info = {
2956                         "action_id": action_id,
2957                         "nsr_id": nsr_id,
2958                         "task_index": task_index,
2959                     }
2960
2961                     target_record = f"{db_record}.{item_index}.vim_info.{target_vim}"
2962                     created_items = self._prepare_created_items_for_healing(
2963                         nsr_id, target_record
2964                     )
2965
2966                     volumes_to_hold = self._prepare_persistent_volumes_for_healing(
2967                         target_vim, existing_instance
2968                     )
2969
2970                     # Specific extra params for recreate tasks:
2971                     extra_dict = {
2972                         "created_items": created_items,
2973                         "vim_id": existing_instance["vim-id"],
2974                         "volumes_to_hold": volumes_to_hold,
2975                     }
2976
2977                     changes_list.append(
2978                         {
2979                             "deployment_info": deployment_info,
2980                             "target_id": target_vim,
2981                             "item": "vdu",
2982                             "action": "DELETE",
2983                             "target_record": target_record,
2984                             "target_record_id": target_record_id,
2985                             "extra_dict": extra_dict,
2986                         }
2987                     )
2988                     delete_task_id = f"{action_id}:{task_index}"
2989                     task_index += 1
2990
2991                     # step 2 vdu to be created
2992                     kwargs = {}
2993                     kwargs.update(
2994                         {
2995                             "vnfr_id": vnfr_id,
2996                             "nsr_id": nsr_id,
2997                             "vnfr": existing_vnf,
2998                             "vdu2cloud_init": vdu2cloud_init,
2999                             "tasks_by_target_record_id": tasks_by_target_record_id,
3000                             "logger": self.logger,
3001                             "db": self.db,
3002                             "fs": self.fs,
3003                             "ro_nsr_public_key": ro_nsr_public_key,
3004                         }
3005                     )
3006
3007                     extra_dict = self._process_recreate_vdu_params(
3008                         existing_instance,
3009                         db_nsr,
3010                         target_viminfo,
3011                         target_record_id,
3012                         target_vim,
3013                         **kwargs,
3014                     )
3015
3016                     # The CREATE task depens on the DELETE task
3017                     extra_dict["depends_on"] = [delete_task_id]
3018
3019                     # Add volumes created from created_items if any
3020                     # Ports should be deleted with delete task and automatically created with create task
3021                     volumes = {}
3022                     for k, v in created_items.items():
3023                         try:
3024                             k_item, _, k_id = k.partition(":")
3025                             if k_item == "volume":
3026                                 volumes[k] = v
3027                         except Exception as e:
3028                             self.logger.error(
3029                                 "Error evaluating created item {}: {}".format(k, e)
3030                             )
3031                     extra_dict["previous_created_volumes"] = volumes
3032
3033                     deployment_info = {
3034                         "action_id": action_id,
3035                         "nsr_id": nsr_id,
3036                         "task_index": task_index,
3037                     }
3038                     self._assign_vim(target_vim)
3039
3040                     new_item = {
3041                         "deployment_info": deployment_info,
3042                         "target_id": target_vim,
3043                         "item": "vdu",
3044                         "action": "CREATE",
3045                         "target_record": target_record,
3046                         "target_record_id": target_record_id,
3047                         "extra_dict": extra_dict,
3048                     }
3049                     changes_list.append(new_item)
3050                     tasks_by_target_record_id[target_record_id] = new_item
3051                     task_index += 1
3052
3053         return changes_list
3054
3055     def recreate(self, session, indata, version, nsr_id, *args, **kwargs):
3056         self.logger.debug("ns.recreate nsr_id={} indata={}".format(nsr_id, indata))
3057         # TODO: validate_input(indata, recreate_schema)
3058         action_id = indata.get("action_id", str(uuid4()))
3059         # get current deployment
3060         db_vnfrs = {}  # vnf's info indexed by _id
3061         step = ""
3062         logging_text = "Recreate nsr_id={} action_id={} indata={}".format(
3063             nsr_id, action_id, indata
3064         )
3065         self.logger.debug(logging_text + "Enter")
3066
3067         try:
3068             step = "Getting ns and vnfr record from db"
3069             db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
3070             db_new_tasks = []
3071             tasks_by_target_record_id = {}
3072             # read from db: vnf's of this ns
3073             step = "Getting vnfrs from db"
3074             db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
3075             self.logger.debug("ns.recreate: db_vnfrs_list={}".format(db_vnfrs_list))
3076
3077             if not db_vnfrs_list:
3078                 raise NsException("Cannot obtain associated VNF for ns")
3079
3080             for vnfr in db_vnfrs_list:
3081                 db_vnfrs[vnfr["_id"]] = vnfr
3082
3083             now = time()
3084             db_ro_nsr = self.db.get_one("ro_nsrs", {"_id": nsr_id}, fail_on_empty=False)
3085             self.logger.debug("ns.recreate: db_ro_nsr={}".format(db_ro_nsr))
3086
3087             if not db_ro_nsr:
3088                 db_ro_nsr = self._create_db_ro_nsrs(nsr_id, now)
3089
3090             with self.write_lock:
3091                 # NS
3092                 step = "process NS elements"
3093                 changes_list = self.prepare_changes_to_recreate(
3094                     indata=indata,
3095                     nsr_id=nsr_id,
3096                     db_nsr=db_nsr,
3097                     db_vnfrs=db_vnfrs,
3098                     db_ro_nsr=db_ro_nsr,
3099                     action_id=action_id,
3100                     tasks_by_target_record_id=tasks_by_target_record_id,
3101                 )
3102
3103                 self.define_all_tasks(
3104                     changes_list=changes_list,
3105                     db_new_tasks=db_new_tasks,
3106                     tasks_by_target_record_id=tasks_by_target_record_id,
3107                 )
3108
3109                 # Delete all ro_tasks registered for the targets vdurs (target_record)
3110                 # If task of type CREATE exist then vim will try to get info form deleted VMs.
3111                 # So remove all task related to target record.
3112                 ro_tasks = self.db.get_list("ro_tasks", {"tasks.nsr_id": nsr_id})
3113                 for change in changes_list:
3114                     for ro_task in ro_tasks:
3115                         for task in ro_task["tasks"]:
3116                             if task["target_record"] == change["target_record"]:
3117                                 self.db.del_one(
3118                                     "ro_tasks",
3119                                     q_filter={
3120                                         "_id": ro_task["_id"],
3121                                         "modified_at": ro_task["modified_at"],
3122                                     },
3123                                     fail_on_empty=False,
3124                                 )
3125
3126                 step = "Updating database, Appending tasks to ro_tasks"
3127                 self.upload_recreate_tasks(
3128                     db_new_tasks=db_new_tasks,
3129                     now=now,
3130                 )
3131
3132             self.logger.debug(
3133                 logging_text + "Exit. Created {} tasks".format(len(db_new_tasks))
3134             )
3135
3136             return (
3137                 {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
3138                 action_id,
3139                 True,
3140             )
3141         except Exception as e:
3142             if isinstance(e, (DbException, NsException)):
3143                 self.logger.error(
3144                     logging_text + "Exit Exception while '{}': {}".format(step, e)
3145                 )
3146             else:
3147                 e = traceback_format_exc()
3148                 self.logger.critical(
3149                     logging_text + "Exit Exception while '{}': {}".format(step, e),
3150                     exc_info=True,
3151                 )
3152
3153             raise NsException(e)
3154
3155     def deploy(self, session, indata, version, nsr_id, *args, **kwargs):
3156         self.logger.debug("ns.deploy nsr_id={} indata={}".format(nsr_id, indata))
3157         validate_input(indata, deploy_schema)
3158         action_id = indata.get("action_id", str(uuid4()))
3159         task_index = 0
3160         # get current deployment
3161         db_nsr_update = {}  # update operation on nsrs
3162         db_vnfrs_update = {}
3163         db_vnfrs = {}  # vnf's info indexed by _id
3164         step = ""
3165         logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id)
3166         self.logger.debug(logging_text + "Enter")
3167
3168         try:
3169             step = "Getting ns and vnfr record from db"
3170             db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
3171             self.logger.debug("ns.deploy: db_nsr={}".format(db_nsr))
3172             db_new_tasks = []
3173             tasks_by_target_record_id = {}
3174             # read from db: vnf's of this ns
3175             step = "Getting vnfrs from db"
3176             db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
3177
3178             if not db_vnfrs_list:
3179                 raise NsException("Cannot obtain associated VNF for ns")
3180
3181             for vnfr in db_vnfrs_list:
3182                 db_vnfrs[vnfr["_id"]] = vnfr
3183                 db_vnfrs_update[vnfr["_id"]] = {}
3184             self.logger.debug("ns.deploy db_vnfrs={}".format(db_vnfrs))
3185
3186             now = time()
3187             db_ro_nsr = self.db.get_one("ro_nsrs", {"_id": nsr_id}, fail_on_empty=False)
3188
3189             if not db_ro_nsr:
3190                 db_ro_nsr = self._create_db_ro_nsrs(nsr_id, now)
3191
3192             # check that action_id is not in the list of actions. Suffixed with :index
3193             if action_id in db_ro_nsr["actions"]:
3194                 index = 1
3195
3196                 while True:
3197                     new_action_id = "{}:{}".format(action_id, index)
3198
3199                     if new_action_id not in db_ro_nsr["actions"]:
3200                         action_id = new_action_id
3201                         self.logger.debug(
3202                             logging_text
3203                             + "Changing action_id in use to {}".format(action_id)
3204                         )
3205                         break
3206
3207                     index += 1
3208
3209             def _process_action(indata):
3210                 nonlocal db_new_tasks
3211                 nonlocal action_id
3212                 nonlocal nsr_id
3213                 nonlocal task_index
3214                 nonlocal db_vnfrs
3215                 nonlocal db_ro_nsr
3216
3217                 if indata["action"]["action"] == "inject_ssh_key":
3218                     key = indata["action"].get("key")
3219                     user = indata["action"].get("user")
3220                     password = indata["action"].get("password")
3221
3222                     for vnf in indata.get("vnf", ()):
3223                         if vnf["_id"] not in db_vnfrs:
3224                             raise NsException("Invalid vnf={}".format(vnf["_id"]))
3225
3226                         db_vnfr = db_vnfrs[vnf["_id"]]
3227
3228                         for target_vdu in vnf.get("vdur", ()):
3229                             vdu_index, vdur = next(
3230                                 (
3231                                     i_v
3232                                     for i_v in enumerate(db_vnfr["vdur"])
3233                                     if i_v[1]["id"] == target_vdu["id"]
3234                                 ),
3235                                 (None, None),
3236                             )
3237
3238                             if not vdur:
3239                                 raise NsException(
3240                                     "Invalid vdu vnf={}.{}".format(
3241                                         vnf["_id"], target_vdu["id"]
3242                                     )
3243                                 )
3244
3245                             target_vim, vim_info = next(
3246                                 k_v for k_v in vdur["vim_info"].items()
3247                             )
3248                             self._assign_vim(target_vim)
3249                             target_record = "vnfrs:{}:vdur.{}.ssh_keys".format(
3250                                 vnf["_id"], vdu_index
3251                             )
3252                             extra_dict = {
3253                                 "depends_on": [
3254                                     "vnfrs:{}:vdur.{}".format(vnf["_id"], vdur["id"])
3255                                 ],
3256                                 "params": {
3257                                     "ip_address": vdur.get("ip-address"),
3258                                     "user": user,
3259                                     "key": key,
3260                                     "password": password,
3261                                     "private_key": db_ro_nsr["private_key"],
3262                                     "salt": db_ro_nsr["_id"],
3263                                     "schema_version": db_ro_nsr["_admin"][
3264                                         "schema_version"
3265                                     ],
3266                                 },
3267                             }
3268
3269                             deployment_info = {
3270                                 "action_id": action_id,
3271                                 "nsr_id": nsr_id,
3272                                 "task_index": task_index,
3273                             }
3274
3275                             task = Ns._create_task(
3276                                 deployment_info=deployment_info,
3277                                 target_id=target_vim,
3278                                 item="vdu",
3279                                 action="EXEC",
3280                                 target_record=target_record,
3281                                 target_record_id=None,
3282                                 extra_dict=extra_dict,
3283                             )
3284
3285                             task_index = deployment_info.get("task_index")
3286
3287                             db_new_tasks.append(task)
3288
3289             with self.write_lock:
3290                 if indata.get("action"):
3291                     _process_action(indata)
3292                 else:
3293                     # compute network differences
3294                     # NS
3295                     step = "process NS elements"
3296                     changes_list = self.calculate_all_differences_to_deploy(
3297                         indata=indata,
3298                         nsr_id=nsr_id,
3299                         db_nsr=db_nsr,
3300                         db_vnfrs=db_vnfrs,
3301                         db_ro_nsr=db_ro_nsr,
3302                         db_nsr_update=db_nsr_update,
3303                         db_vnfrs_update=db_vnfrs_update,
3304                         action_id=action_id,
3305                         tasks_by_target_record_id=tasks_by_target_record_id,
3306                     )
3307                     self.define_all_tasks(
3308                         changes_list=changes_list,
3309                         db_new_tasks=db_new_tasks,
3310                         tasks_by_target_record_id=tasks_by_target_record_id,
3311                     )
3312
3313                 step = "Updating database, Appending tasks to ro_tasks"
3314                 self.upload_all_tasks(
3315                     db_new_tasks=db_new_tasks,
3316                     now=now,
3317                 )
3318
3319                 step = "Updating database, nsrs"
3320                 if db_nsr_update:
3321                     self.db.set_one("nsrs", {"_id": nsr_id}, db_nsr_update)
3322
3323                 for vnfr_id, db_vnfr_update in db_vnfrs_update.items():
3324                     if db_vnfr_update:
3325                         step = "Updating database, vnfrs={}".format(vnfr_id)
3326                         self.db.set_one("vnfrs", {"_id": vnfr_id}, db_vnfr_update)
3327
3328             self.logger.debug(
3329                 logging_text + "Exit. Created {} tasks".format(len(db_new_tasks))
3330             )
3331
3332             return (
3333                 {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
3334                 action_id,
3335                 True,
3336             )
3337         except Exception as e:
3338             if isinstance(e, (DbException, NsException)):
3339                 self.logger.error(
3340                     logging_text + "Exit Exception while '{}': {}".format(step, e)
3341                 )
3342             else:
3343                 e = traceback_format_exc()
3344                 self.logger.critical(
3345                     logging_text + "Exit Exception while '{}': {}".format(step, e),
3346                     exc_info=True,
3347                 )
3348
3349             raise NsException(e)
3350
3351     def delete(self, session, indata, version, nsr_id, *args, **kwargs):
3352         self.logger.debug("ns.delete version={} nsr_id={}".format(version, nsr_id))
3353         # self.db.del_list({"_id": ro_task["_id"], "tasks.nsr_id.ne": nsr_id})
3354
3355         with self.write_lock:
3356             try:
3357                 NsWorker.delete_db_tasks(self.db, nsr_id, None)
3358             except NsWorkerException as e:
3359                 raise NsException(e)
3360
3361         return None, None, True
3362
3363     def status(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
3364         self.logger.debug(
3365             "ns.status version={} nsr_id={}, action_id={} indata={}".format(
3366                 version, nsr_id, action_id, indata
3367             )
3368         )
3369         task_list = []
3370         done = 0
3371         total = 0
3372         ro_tasks = self.db.get_list("ro_tasks", {"tasks.action_id": action_id})
3373         global_status = "DONE"
3374         details = []
3375
3376         for ro_task in ro_tasks:
3377             for task in ro_task["tasks"]:
3378                 if task and task["action_id"] == action_id:
3379                     task_list.append(task)
3380                     total += 1
3381
3382                     if task["status"] == "FAILED":
3383                         global_status = "FAILED"
3384                         error_text = "Error at {} {}: {}".format(
3385                             task["action"].lower(),
3386                             task["item"],
3387                             ro_task["vim_info"].get("vim_message") or "unknown",
3388                         )
3389                         details.append(error_text)
3390                     elif task["status"] in ("SCHEDULED", "BUILD"):
3391                         if global_status != "FAILED":
3392                             global_status = "BUILD"
3393                     else:
3394                         done += 1
3395
3396         return_data = {
3397             "status": global_status,
3398             "details": (
3399                 ". ".join(details) if details else "progress {}/{}".format(done, total)
3400             ),
3401             "nsr_id": nsr_id,
3402             "action_id": action_id,
3403             "tasks": task_list,
3404         }
3405
3406         return return_data, None, True
3407
3408     def recreate_status(
3409         self, session, indata, version, nsr_id, action_id, *args, **kwargs
3410     ):
3411         return self.status(session, indata, version, nsr_id, action_id, *args, **kwargs)
3412
3413     def cancel(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
3414         print(
3415             "ns.cancel session={} indata={} version={} nsr_id={}, action_id={}".format(
3416                 session, indata, version, nsr_id, action_id
3417             )
3418         )
3419
3420         return None, None, True
3421
3422     def rebuild_start_stop_task(
3423         self,
3424         vdu_id,
3425         vnf_id,
3426         vdu_index,
3427         action_id,
3428         nsr_id,
3429         task_index,
3430         target_vim,
3431         extra_dict,
3432     ):
3433         self._assign_vim(target_vim)
3434         target_record = "vnfrs:{}:vdur.{}.vim_info.{}".format(
3435             vnf_id, vdu_index, target_vim
3436         )
3437         target_record_id = "vnfrs:{}:vdur.{}".format(vnf_id, vdu_id)
3438         deployment_info = {
3439             "action_id": action_id,
3440             "nsr_id": nsr_id,
3441             "task_index": task_index,
3442         }
3443
3444         task = Ns._create_task(
3445             deployment_info=deployment_info,
3446             target_id=target_vim,
3447             item="update",
3448             action="EXEC",
3449             target_record=target_record,
3450             target_record_id=target_record_id,
3451             extra_dict=extra_dict,
3452         )
3453         return task
3454
3455     def rebuild_start_stop(
3456         self, session, action_dict, version, nsr_id, *args, **kwargs
3457     ):
3458         task_index = 0
3459         extra_dict = {}
3460         now = time()
3461         action_id = action_dict.get("action_id", str(uuid4()))
3462         step = ""
3463         logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id)
3464         self.logger.debug(logging_text + "Enter")
3465
3466         action = list(action_dict.keys())[0]
3467         task_dict = action_dict.get(action)
3468         vim_vm_id = action_dict.get(action).get("vim_vm_id")
3469
3470         if action_dict.get("stop"):
3471             action = "shutoff"
3472         db_new_tasks = []
3473         try:
3474             step = "lock the operation & do task creation"
3475             with self.write_lock:
3476                 extra_dict["params"] = {
3477                     "vim_vm_id": vim_vm_id,
3478                     "action": action,
3479                 }
3480                 task = self.rebuild_start_stop_task(
3481                     task_dict["vdu_id"],
3482                     task_dict["vnf_id"],
3483                     task_dict["vdu_index"],
3484                     action_id,
3485                     nsr_id,
3486                     task_index,
3487                     task_dict["target_vim"],
3488                     extra_dict,
3489                 )
3490                 db_new_tasks.append(task)
3491                 step = "upload Task to db"
3492                 self.upload_all_tasks(
3493                     db_new_tasks=db_new_tasks,
3494                     now=now,
3495                 )
3496                 self.logger.debug(
3497                     logging_text + "Exit. Created {} tasks".format(len(db_new_tasks))
3498                 )
3499                 return (
3500                     {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
3501                     action_id,
3502                     True,
3503                 )
3504         except Exception as e:
3505             if isinstance(e, (DbException, NsException)):
3506                 self.logger.error(
3507                     logging_text + "Exit Exception while '{}': {}".format(step, e)
3508                 )
3509             else:
3510                 e = traceback_format_exc()
3511                 self.logger.critical(
3512                     logging_text + "Exit Exception while '{}': {}".format(step, e),
3513                     exc_info=True,
3514                 )
3515             raise NsException(e)
3516
3517     def get_deploy(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
3518         nsrs = self.db.get_list("nsrs", {})
3519         return_data = []
3520
3521         for ns in nsrs:
3522             return_data.append({"_id": ns["_id"], "name": ns["name"]})
3523
3524         return return_data, None, True
3525
3526     def get_actions(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
3527         ro_tasks = self.db.get_list("ro_tasks", {"tasks.nsr_id": nsr_id})
3528         return_data = []
3529
3530         for ro_task in ro_tasks:
3531             for task in ro_task["tasks"]:
3532                 if task["action_id"] not in return_data:
3533                     return_data.append(task["action_id"])
3534
3535         return return_data, None, True
3536
3537     def migrate_task(
3538         self, vdu, vnf, vdu_index, action_id, nsr_id, task_index, extra_dict
3539     ):
3540         target_vim, vim_info = next(k_v for k_v in vdu["vim_info"].items())
3541         self._assign_vim(target_vim)
3542         target_record = "vnfrs:{}:vdur.{}.vim_info.{}".format(
3543             vnf["_id"], vdu_index, target_vim
3544         )
3545         target_record_id = "vnfrs:{}:vdur.{}".format(vnf["_id"], vdu["id"])
3546         deployment_info = {
3547             "action_id": action_id,
3548             "nsr_id": nsr_id,
3549             "task_index": task_index,
3550         }
3551
3552         task = Ns._create_task(
3553             deployment_info=deployment_info,
3554             target_id=target_vim,
3555             item="migrate",
3556             action="EXEC",
3557             target_record=target_record,
3558             target_record_id=target_record_id,
3559             extra_dict=extra_dict,
3560         )
3561
3562         return task
3563
3564     def migrate(self, session, indata, version, nsr_id, *args, **kwargs):
3565         task_index = 0
3566         extra_dict = {}
3567         now = time()
3568         action_id = indata.get("action_id", str(uuid4()))
3569         step = ""
3570         logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id)
3571         self.logger.debug(logging_text + "Enter")
3572         try:
3573             vnf_instance_id = indata["vnfInstanceId"]
3574             step = "Getting vnfrs from db"
3575             db_vnfr = self.db.get_one("vnfrs", {"_id": vnf_instance_id})
3576             vdu = indata.get("vdu")
3577             migrateToHost = indata.get("migrateToHost")
3578             db_new_tasks = []
3579
3580             with self.write_lock:
3581                 if vdu is not None:
3582                     vdu_id = indata["vdu"]["vduId"]
3583                     vdu_count_index = indata["vdu"].get("vduCountIndex", 0)
3584                     for vdu_index, vdu in enumerate(db_vnfr["vdur"]):
3585                         if (
3586                             vdu["vdu-id-ref"] == vdu_id
3587                             and vdu["count-index"] == vdu_count_index
3588                         ):
3589                             extra_dict["params"] = {
3590                                 "vim_vm_id": vdu["vim-id"],
3591                                 "migrate_host": migrateToHost,
3592                                 "vdu_vim_info": vdu["vim_info"],
3593                             }
3594                             step = "Creating migration task for vdu:{}".format(vdu)
3595                             task = self.migrate_task(
3596                                 vdu,
3597                                 db_vnfr,
3598                                 vdu_index,
3599                                 action_id,
3600                                 nsr_id,
3601                                 task_index,
3602                                 extra_dict,
3603                             )
3604                             db_new_tasks.append(task)
3605                             task_index += 1
3606                             break
3607                 else:
3608                     for vdu_index, vdu in enumerate(db_vnfr["vdur"]):
3609                         extra_dict["params"] = {
3610                             "vim_vm_id": vdu["vim-id"],
3611                             "migrate_host": migrateToHost,
3612                             "vdu_vim_info": vdu["vim_info"],
3613                         }
3614                         step = "Creating migration task for vdu:{}".format(vdu)
3615                         task = self.migrate_task(
3616                             vdu,
3617                             db_vnfr,
3618                             vdu_index,
3619                             action_id,
3620                             nsr_id,
3621                             task_index,
3622                             extra_dict,
3623                         )
3624                         db_new_tasks.append(task)
3625                         task_index += 1
3626
3627                 self.upload_all_tasks(
3628                     db_new_tasks=db_new_tasks,
3629                     now=now,
3630                 )
3631
3632             self.logger.debug(
3633                 logging_text + "Exit. Created {} tasks".format(len(db_new_tasks))
3634             )
3635             return (
3636                 {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
3637                 action_id,
3638                 True,
3639             )
3640         except Exception as e:
3641             if isinstance(e, (DbException, NsException)):
3642                 self.logger.error(
3643                     logging_text + "Exit Exception while '{}': {}".format(step, e)
3644                 )
3645             else:
3646                 e = traceback_format_exc()
3647                 self.logger.critical(
3648                     logging_text + "Exit Exception while '{}': {}".format(step, e),
3649                     exc_info=True,
3650                 )
3651             raise NsException(e)
3652
3653     def verticalscale_task(
3654         self, vdu, vnf, vdu_index, action_id, nsr_id, task_index, extra_dict
3655     ):
3656         target_vim, vim_info = next(k_v for k_v in vdu["vim_info"].items())
3657         self._assign_vim(target_vim)
3658         ns_preffix = "nsrs:{}".format(nsr_id)
3659         flavor_text = ns_preffix + ":flavor." + vdu["ns-flavor-id"]
3660         extra_dict["depends_on"] = [flavor_text]
3661         extra_dict["params"].update({"flavor_id": "TASK-" + flavor_text})
3662         target_record = "vnfrs:{}:vdur.{}.vim_info.{}".format(
3663             vnf["_id"], vdu_index, target_vim
3664         )
3665         target_record_id = "vnfrs:{}:vdur.{}".format(vnf["_id"], vdu["id"])
3666         deployment_info = {
3667             "action_id": action_id,
3668             "nsr_id": nsr_id,
3669             "task_index": task_index,
3670         }
3671
3672         task = Ns._create_task(
3673             deployment_info=deployment_info,
3674             target_id=target_vim,
3675             item="verticalscale",
3676             action="EXEC",
3677             target_record=target_record,
3678             target_record_id=target_record_id,
3679             extra_dict=extra_dict,
3680         )
3681         return task
3682
3683     def verticalscale_flavor_task(
3684         self, vdu, vnf, vdu_index, action_id, nsr_id, task_index, extra_dict
3685     ):
3686         target_vim, vim_info = next(k_v for k_v in vdu["vim_info"].items())
3687         self._assign_vim(target_vim)
3688         db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
3689         target_record = "nsrs:{}:flavor.{}.vim_info.{}".format(
3690             nsr_id, len(db_nsr["flavor"]) - 1, target_vim
3691         )
3692         target_record_id = "nsrs:{}:flavor.{}".format(nsr_id, len(db_nsr["flavor"]) - 1)
3693         deployment_info = {
3694             "action_id": action_id,
3695             "nsr_id": nsr_id,
3696             "task_index": task_index,
3697         }
3698         task = Ns._create_task(
3699             deployment_info=deployment_info,
3700             target_id=target_vim,
3701             item="flavor",
3702             action="CREATE",
3703             target_record=target_record,
3704             target_record_id=target_record_id,
3705             extra_dict=extra_dict,
3706         )
3707         return task
3708
3709     def verticalscale(self, session, indata, version, nsr_id, *args, **kwargs):
3710         task_index = 0
3711         extra_dict = {}
3712         flavor_extra_dict = {}
3713         now = time()
3714         action_id = indata.get("action_id", str(uuid4()))
3715         step = ""
3716         logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id)
3717         self.logger.debug(logging_text + "Enter")
3718         try:
3719             VnfFlavorData = indata.get("changeVnfFlavorData")
3720             vnf_instance_id = VnfFlavorData["vnfInstanceId"]
3721             step = "Getting vnfrs from db"
3722             db_vnfr = self.db.get_one("vnfrs", {"_id": vnf_instance_id})
3723             vduid = VnfFlavorData["additionalParams"]["vduid"]
3724             vduCountIndex = VnfFlavorData["additionalParams"]["vduCountIndex"]
3725             virtualMemory = VnfFlavorData["additionalParams"]["virtualMemory"]
3726             numVirtualCpu = VnfFlavorData["additionalParams"]["numVirtualCpu"]
3727             sizeOfStorage = VnfFlavorData["additionalParams"]["sizeOfStorage"]
3728             flavor_dict = {
3729                 "name": vduid + "-flv",
3730                 "ram": virtualMemory,
3731                 "vcpus": numVirtualCpu,
3732                 "disk": sizeOfStorage,
3733             }
3734             flavor_data = {
3735                 "ram": virtualMemory,
3736                 "vcpus": numVirtualCpu,
3737                 "disk": sizeOfStorage,
3738             }
3739             flavor_extra_dict["find_params"] = {"flavor_data": flavor_data}
3740             flavor_extra_dict["params"] = {"flavor_data": flavor_dict}
3741             db_new_tasks = []
3742             step = "Creating Tasks for vertical scaling"
3743             with self.write_lock:
3744                 for vdu_index, vdu in enumerate(db_vnfr["vdur"]):
3745                     if (
3746                         vdu["vdu-id-ref"] == vduid
3747                         and vdu["count-index"] == vduCountIndex
3748                     ):
3749                         extra_dict["params"] = {
3750                             "vim_vm_id": vdu["vim-id"],
3751                             "flavor_dict": flavor_dict,
3752                             "vdu-id-ref": vdu["vdu-id-ref"],
3753                             "count-index": vdu["count-index"],
3754                             "vnf_instance_id": vnf_instance_id,
3755                         }
3756                         task = self.verticalscale_flavor_task(
3757                             vdu,
3758                             db_vnfr,
3759                             vdu_index,
3760                             action_id,
3761                             nsr_id,
3762                             task_index,
3763                             flavor_extra_dict,
3764                         )
3765                         db_new_tasks.append(task)
3766                         task_index += 1
3767                         task = self.verticalscale_task(
3768                             vdu,
3769                             db_vnfr,
3770                             vdu_index,
3771                             action_id,
3772                             nsr_id,
3773                             task_index,
3774                             extra_dict,
3775                         )
3776                         db_new_tasks.append(task)
3777                         task_index += 1
3778                         break
3779                 self.upload_all_tasks(
3780                     db_new_tasks=db_new_tasks,
3781                     now=now,
3782                 )
3783             self.logger.debug(
3784                 logging_text + "Exit. Created {} tasks".format(len(db_new_tasks))
3785             )
3786             return (
3787                 {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
3788                 action_id,
3789                 True,
3790             )
3791         except Exception as e:
3792             if isinstance(e, (DbException, NsException)):
3793                 self.logger.error(
3794                     logging_text + "Exit Exception while '{}': {}".format(step, e)
3795                 )
3796             else:
3797                 e = traceback_format_exc()
3798                 self.logger.critical(
3799                     logging_text + "Exit Exception while '{}': {}".format(step, e),
3800                     exc_info=True,
3801                 )
3802             raise NsException(e)