NG-RO/osm_ng_ro/ns.py

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