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
2318             vnfr_id = target_vnf.get("vnfInstanceId", None)
2319             if vnfr_id:
2320                 existing_vnf = db_vnfrs.get(vnfr_id)
2321                 db_record = "vnfrs:{}:{}".format(vnfr_id, db_path)
2322                 # vim_account_id = existing_vnf.get("vim-account-id", "")
2323
2324             # Check each VDU of this VNF
2325             for target_vdu in target_vnf["additionalParams"].get("vdu", None):
2326                 vdu_name = target_vdu.get("vdu-id", None)
2327                 # For multi instance VDU count-index is mandatory
2328                 # For single session VDU count-indes is 0
2329                 count_index = target_vdu.get("count-index", 0)
2330                 item_index = 0
2331                 existing_instance = None
2332                 for instance in existing_vnf.get("vdur", None):
2333                     if (
2334                         instance["vdu-name"] == vdu_name
2335                         and instance["count-index"] == count_index
2336                     ):
2337                         existing_instance = instance
2338                         break
2339                     else:
2340                         item_index += 1
2341
2342                 target_record_id = "{}.{}".format(db_record, existing_instance["id"])
2343
2344                 # The target VIM is the one already existing in DB to recreate
2345                 for target_vim, target_viminfo in existing_instance.get(
2346                     "vim_info", {}
2347                 ).items():
2348                     # step 1 vdu to be deleted
2349                     self._assign_vim(target_vim)
2350                     deployment_info = {
2351                         "action_id": action_id,
2352                         "nsr_id": nsr_id,
2353                         "task_index": task_index,
2354                     }
2355
2356                     target_record = f"{db_record}.{item_index}.vim_info.{target_vim}"
2357                     created_items = self._prepare_created_items_for_healing(
2358                         nsr_id, target_record
2359                     )
2360
2361                     volumes_to_hold = self._prepare_persistent_volumes_for_healing(
2362                         target_vim, existing_instance
2363                     )
2364
2365                     # Specific extra params for recreate tasks:
2366                     extra_dict = {
2367                         "created_items": created_items,
2368                         "vim_id": existing_instance["vim-id"],
2369                         "volumes_to_hold": volumes_to_hold,
2370                     }
2371
2372                     changes_list.append(
2373                         {
2374                             "deployment_info": deployment_info,
2375                             "target_id": target_vim,
2376                             "item": "vdu",
2377                             "action": "DELETE",
2378                             "target_record": target_record,
2379                             "target_record_id": target_record_id,
2380                             "extra_dict": extra_dict,
2381                         }
2382                     )
2383                     delete_task_id = f"{action_id}:{task_index}"
2384                     task_index += 1
2385
2386                     # step 2 vdu to be created
2387                     kwargs = {}
2388                     kwargs.update(
2389                         {
2390                             "vnfr_id": vnfr_id,
2391                             "nsr_id": nsr_id,
2392                             "vnfr": existing_vnf,
2393                             "vdu2cloud_init": vdu2cloud_init,
2394                             "tasks_by_target_record_id": tasks_by_target_record_id,
2395                             "logger": self.logger,
2396                             "db": self.db,
2397                             "fs": self.fs,
2398                             "ro_nsr_public_key": ro_nsr_public_key,
2399                         }
2400                     )
2401
2402                     extra_dict = self._process_recreate_vdu_params(
2403                         existing_instance,
2404                         db_nsr,
2405                         target_viminfo,
2406                         target_record_id,
2407                         target_vim,
2408                         **kwargs,
2409                     )
2410
2411                     # The CREATE task depens on the DELETE task
2412                     extra_dict["depends_on"] = [delete_task_id]
2413
2414                     # Add volumes created from created_items if any
2415                     # Ports should be deleted with delete task and automatically created with create task
2416                     volumes = {}
2417                     for k, v in created_items.items():
2418                         try:
2419                             k_item, _, k_id = k.partition(":")
2420                             if k_item == "volume":
2421                                 volumes[k] = v
2422                         except Exception as e:
2423                             self.logger.error(
2424                                 "Error evaluating created item {}: {}".format(k, e)
2425                             )
2426                     extra_dict["previous_created_volumes"] = volumes
2427
2428                     deployment_info = {
2429                         "action_id": action_id,
2430                         "nsr_id": nsr_id,
2431                         "task_index": task_index,
2432                     }
2433                     self._assign_vim(target_vim)
2434
2435                     new_item = {
2436                         "deployment_info": deployment_info,
2437                         "target_id": target_vim,
2438                         "item": "vdu",
2439                         "action": "CREATE",
2440                         "target_record": target_record,
2441                         "target_record_id": target_record_id,
2442                         "extra_dict": extra_dict,
2443                     }
2444                     changes_list.append(new_item)
2445                     tasks_by_target_record_id[target_record_id] = new_item
2446                     task_index += 1
2447
2448         return changes_list
2449
2450     def recreate(self, session, indata, version, nsr_id, *args, **kwargs):
2451         self.logger.debug("ns.recreate nsr_id={} indata={}".format(nsr_id, indata))
2452         # TODO: validate_input(indata, recreate_schema)
2453         action_id = indata.get("action_id", str(uuid4()))
2454         # get current deployment
2455         db_vnfrs = {}  # vnf's info indexed by _id
2456         step = ""
2457         logging_text = "Recreate nsr_id={} action_id={} indata={}".format(
2458             nsr_id, action_id, indata
2459         )
2460         self.logger.debug(logging_text + "Enter")
2461
2462         try:
2463             step = "Getting ns and vnfr record from db"
2464             db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
2465             db_new_tasks = []
2466             tasks_by_target_record_id = {}
2467             # read from db: vnf's of this ns
2468             step = "Getting vnfrs from db"
2469             db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
2470             self.logger.debug("ns.recreate: db_vnfrs_list={}".format(db_vnfrs_list))
2471
2472             if not db_vnfrs_list:
2473                 raise NsException("Cannot obtain associated VNF for ns")
2474
2475             for vnfr in db_vnfrs_list:
2476                 db_vnfrs[vnfr["_id"]] = vnfr
2477
2478             now = time()
2479             db_ro_nsr = self.db.get_one("ro_nsrs", {"_id": nsr_id}, fail_on_empty=False)
2480             self.logger.debug("ns.recreate: db_ro_nsr={}".format(db_ro_nsr))
2481
2482             if not db_ro_nsr:
2483                 db_ro_nsr = self._create_db_ro_nsrs(nsr_id, now)
2484
2485             with self.write_lock:
2486                 # NS
2487                 step = "process NS elements"
2488                 changes_list = self.prepare_changes_to_recreate(
2489                     indata=indata,
2490                     nsr_id=nsr_id,
2491                     db_nsr=db_nsr,
2492                     db_vnfrs=db_vnfrs,
2493                     db_ro_nsr=db_ro_nsr,
2494                     action_id=action_id,
2495                     tasks_by_target_record_id=tasks_by_target_record_id,
2496                 )
2497
2498                 self.define_all_tasks(
2499                     changes_list=changes_list,
2500                     db_new_tasks=db_new_tasks,
2501                     tasks_by_target_record_id=tasks_by_target_record_id,
2502                 )
2503
2504                 # Delete all ro_tasks registered for the targets vdurs (target_record)
2505                 # If task of type CREATE exist then vim will try to get info form deleted VMs.
2506                 # So remove all task related to target record.
2507                 ro_tasks = self.db.get_list("ro_tasks", {"tasks.nsr_id": nsr_id})
2508                 for change in changes_list:
2509                     for ro_task in ro_tasks:
2510                         for task in ro_task["tasks"]:
2511                             if task["target_record"] == change["target_record"]:
2512                                 self.db.del_one(
2513                                     "ro_tasks",
2514                                     q_filter={
2515                                         "_id": ro_task["_id"],
2516                                         "modified_at": ro_task["modified_at"],
2517                                     },
2518                                     fail_on_empty=False,
2519                                 )
2520
2521                 step = "Updating database, Appending tasks to ro_tasks"
2522                 self.upload_recreate_tasks(
2523                     db_new_tasks=db_new_tasks,
2524                     now=now,
2525                 )
2526
2527             self.logger.debug(
2528                 logging_text + "Exit. Created {} tasks".format(len(db_new_tasks))
2529             )
2530
2531             return (
2532                 {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
2533                 action_id,
2534                 True,
2535             )
2536         except Exception as e:
2537             if isinstance(e, (DbException, NsException)):
2538                 self.logger.error(
2539                     logging_text + "Exit Exception while '{}': {}".format(step, e)
2540                 )
2541             else:
2542                 e = traceback_format_exc()
2543                 self.logger.critical(
2544                     logging_text + "Exit Exception while '{}': {}".format(step, e),
2545                     exc_info=True,
2546                 )
2547
2548             raise NsException(e)
2549
2550     def deploy(self, session, indata, version, nsr_id, *args, **kwargs):
2551         self.logger.debug("ns.deploy nsr_id={} indata={}".format(nsr_id, indata))
2552         validate_input(indata, deploy_schema)
2553         action_id = indata.get("action_id", str(uuid4()))
2554         task_index = 0
2555         # get current deployment
2556         db_nsr_update = {}  # update operation on nsrs
2557         db_vnfrs_update = {}
2558         db_vnfrs = {}  # vnf's info indexed by _id
2559         step = ""
2560         logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id)
2561         self.logger.debug(logging_text + "Enter")
2562
2563         try:
2564             step = "Getting ns and vnfr record from db"
2565             db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
2566             self.logger.debug("ns.deploy: db_nsr={}".format(db_nsr))
2567             db_new_tasks = []
2568             tasks_by_target_record_id = {}
2569             # read from db: vnf's of this ns
2570             step = "Getting vnfrs from db"
2571             db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
2572
2573             if not db_vnfrs_list:
2574                 raise NsException("Cannot obtain associated VNF for ns")
2575
2576             for vnfr in db_vnfrs_list:
2577                 db_vnfrs[vnfr["_id"]] = vnfr
2578                 db_vnfrs_update[vnfr["_id"]] = {}
2579             self.logger.debug("ns.deploy db_vnfrs={}".format(db_vnfrs))
2580
2581             now = time()
2582             db_ro_nsr = self.db.get_one("ro_nsrs", {"_id": nsr_id}, fail_on_empty=False)
2583
2584             if not db_ro_nsr:
2585                 db_ro_nsr = self._create_db_ro_nsrs(nsr_id, now)
2586
2587             # check that action_id is not in the list of actions. Suffixed with :index
2588             if action_id in db_ro_nsr["actions"]:
2589                 index = 1
2590
2591                 while True:
2592                     new_action_id = "{}:{}".format(action_id, index)
2593
2594                     if new_action_id not in db_ro_nsr["actions"]:
2595                         action_id = new_action_id
2596                         self.logger.debug(
2597                             logging_text
2598                             + "Changing action_id in use to {}".format(action_id)
2599                         )
2600                         break
2601
2602                     index += 1
2603
2604             def _process_action(indata):
2605                 nonlocal db_new_tasks
2606                 nonlocal action_id
2607                 nonlocal nsr_id
2608                 nonlocal task_index
2609                 nonlocal db_vnfrs
2610                 nonlocal db_ro_nsr
2611
2612                 if indata["action"]["action"] == "inject_ssh_key":
2613                     key = indata["action"].get("key")
2614                     user = indata["action"].get("user")
2615                     password = indata["action"].get("password")
2616
2617                     for vnf in indata.get("vnf", ()):
2618                         if vnf["_id"] not in db_vnfrs:
2619                             raise NsException("Invalid vnf={}".format(vnf["_id"]))
2620
2621                         db_vnfr = db_vnfrs[vnf["_id"]]
2622
2623                         for target_vdu in vnf.get("vdur", ()):
2624                             vdu_index, vdur = next(
2625                                 (
2626                                     i_v
2627                                     for i_v in enumerate(db_vnfr["vdur"])
2628                                     if i_v[1]["id"] == target_vdu["id"]
2629                                 ),
2630                                 (None, None),
2631                             )
2632
2633                             if not vdur:
2634                                 raise NsException(
2635                                     "Invalid vdu vnf={}.{}".format(
2636                                         vnf["_id"], target_vdu["id"]
2637                                     )
2638                                 )
2639
2640                             target_vim, vim_info = next(
2641                                 k_v for k_v in vdur["vim_info"].items()
2642                             )
2643                             self._assign_vim(target_vim)
2644                             target_record = "vnfrs:{}:vdur.{}.ssh_keys".format(
2645                                 vnf["_id"], vdu_index
2646                             )
2647                             extra_dict = {
2648                                 "depends_on": [
2649                                     "vnfrs:{}:vdur.{}".format(vnf["_id"], vdur["id"])
2650                                 ],
2651                                 "params": {
2652                                     "ip_address": vdur.get("ip-address"),
2653                                     "user": user,
2654                                     "key": key,
2655                                     "password": password,
2656                                     "private_key": db_ro_nsr["private_key"],
2657                                     "salt": db_ro_nsr["_id"],
2658                                     "schema_version": db_ro_nsr["_admin"][
2659                                         "schema_version"
2660                                     ],
2661                                 },
2662                             }
2663
2664                             deployment_info = {
2665                                 "action_id": action_id,
2666                                 "nsr_id": nsr_id,
2667                                 "task_index": task_index,
2668                             }
2669
2670                             task = Ns._create_task(
2671                                 deployment_info=deployment_info,
2672                                 target_id=target_vim,
2673                                 item="vdu",
2674                                 action="EXEC",
2675                                 target_record=target_record,
2676                                 target_record_id=None,
2677                                 extra_dict=extra_dict,
2678                             )
2679
2680                             task_index = deployment_info.get("task_index")
2681
2682                             db_new_tasks.append(task)
2683
2684             with self.write_lock:
2685                 if indata.get("action"):
2686                     _process_action(indata)
2687                 else:
2688                     # compute network differences
2689                     # NS
2690                     step = "process NS elements"
2691                     changes_list = self.calculate_all_differences_to_deploy(
2692                         indata=indata,
2693                         nsr_id=nsr_id,
2694                         db_nsr=db_nsr,
2695                         db_vnfrs=db_vnfrs,
2696                         db_ro_nsr=db_ro_nsr,
2697                         db_nsr_update=db_nsr_update,
2698                         db_vnfrs_update=db_vnfrs_update,
2699                         action_id=action_id,
2700                         tasks_by_target_record_id=tasks_by_target_record_id,
2701                     )
2702                     self.define_all_tasks(
2703                         changes_list=changes_list,
2704                         db_new_tasks=db_new_tasks,
2705                         tasks_by_target_record_id=tasks_by_target_record_id,
2706                     )
2707
2708                 step = "Updating database, Appending tasks to ro_tasks"
2709                 self.upload_all_tasks(
2710                     db_new_tasks=db_new_tasks,
2711                     now=now,
2712                 )
2713
2714                 step = "Updating database, nsrs"
2715                 if db_nsr_update:
2716                     self.db.set_one("nsrs", {"_id": nsr_id}, db_nsr_update)
2717
2718                 for vnfr_id, db_vnfr_update in db_vnfrs_update.items():
2719                     if db_vnfr_update:
2720                         step = "Updating database, vnfrs={}".format(vnfr_id)
2721                         self.db.set_one("vnfrs", {"_id": vnfr_id}, db_vnfr_update)
2722
2723             self.logger.debug(
2724                 logging_text + "Exit. Created {} tasks".format(len(db_new_tasks))
2725             )
2726
2727             return (
2728                 {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
2729                 action_id,
2730                 True,
2731             )
2732         except Exception as e:
2733             if isinstance(e, (DbException, NsException)):
2734                 self.logger.error(
2735                     logging_text + "Exit Exception while '{}': {}".format(step, e)
2736                 )
2737             else:
2738                 e = traceback_format_exc()
2739                 self.logger.critical(
2740                     logging_text + "Exit Exception while '{}': {}".format(step, e),
2741                     exc_info=True,
2742                 )
2743
2744             raise NsException(e)
2745
2746     def delete(self, session, indata, version, nsr_id, *args, **kwargs):
2747         self.logger.debug("ns.delete version={} nsr_id={}".format(version, nsr_id))
2748         # self.db.del_list({"_id": ro_task["_id"], "tasks.nsr_id.ne": nsr_id})
2749
2750         with self.write_lock:
2751             try:
2752                 NsWorker.delete_db_tasks(self.db, nsr_id, None)
2753             except NsWorkerException as e:
2754                 raise NsException(e)
2755
2756         return None, None, True
2757
2758     def status(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
2759         self.logger.debug(
2760             "ns.status version={} nsr_id={}, action_id={} indata={}".format(
2761                 version, nsr_id, action_id, indata
2762             )
2763         )
2764         task_list = []
2765         done = 0
2766         total = 0
2767         ro_tasks = self.db.get_list("ro_tasks", {"tasks.action_id": action_id})
2768         global_status = "DONE"
2769         details = []
2770
2771         for ro_task in ro_tasks:
2772             for task in ro_task["tasks"]:
2773                 if task and task["action_id"] == action_id:
2774                     task_list.append(task)
2775                     total += 1
2776
2777                     if task["status"] == "FAILED":
2778                         global_status = "FAILED"
2779                         error_text = "Error at {} {}: {}".format(
2780                             task["action"].lower(),
2781                             task["item"],
2782                             ro_task["vim_info"].get("vim_message") or "unknown",
2783                         )
2784                         details.append(error_text)
2785                     elif task["status"] in ("SCHEDULED", "BUILD"):
2786                         if global_status != "FAILED":
2787                             global_status = "BUILD"
2788                     else:
2789                         done += 1
2790
2791         return_data = {
2792             "status": global_status,
2793             "details": ". ".join(details)
2794             if details
2795             else "progress {}/{}".format(done, total),
2796             "nsr_id": nsr_id,
2797             "action_id": action_id,
2798             "tasks": task_list,
2799         }
2800
2801         return return_data, None, True
2802
2803     def recreate_status(
2804         self, session, indata, version, nsr_id, action_id, *args, **kwargs
2805     ):
2806         return self.status(session, indata, version, nsr_id, action_id, *args, **kwargs)
2807
2808     def cancel(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
2809         print(
2810             "ns.cancel session={} indata={} version={} nsr_id={}, action_id={}".format(
2811                 session, indata, version, nsr_id, action_id
2812             )
2813         )
2814
2815         return None, None, True
2816
2817     def rebuild_start_stop_task(
2818         self,
2819         vdu_id,
2820         vnf_id,
2821         vdu_index,
2822         action_id,
2823         nsr_id,
2824         task_index,
2825         target_vim,
2826         extra_dict,
2827     ):
2828         self._assign_vim(target_vim)
2829         target_record = "vnfrs:{}:vdur.{}".format(vnf_id, vdu_index)
2830         target_record_id = "vnfrs:{}:vdur.{}".format(vnf_id, vdu_id)
2831         deployment_info = {
2832             "action_id": action_id,
2833             "nsr_id": nsr_id,
2834             "task_index": task_index,
2835         }
2836
2837         task = Ns._create_task(
2838             deployment_info=deployment_info,
2839             target_id=target_vim,
2840             item="update",
2841             action="EXEC",
2842             target_record=target_record,
2843             target_record_id=target_record_id,
2844             extra_dict=extra_dict,
2845         )
2846         return task
2847
2848     def rebuild_start_stop(
2849         self, session, action_dict, version, nsr_id, *args, **kwargs
2850     ):
2851         task_index = 0
2852         extra_dict = {}
2853         now = time()
2854         action_id = action_dict.get("action_id", str(uuid4()))
2855         step = ""
2856         logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id)
2857         self.logger.debug(logging_text + "Enter")
2858
2859         action = list(action_dict.keys())[0]
2860         task_dict = action_dict.get(action)
2861         vim_vm_id = action_dict.get(action).get("vim_vm_id")
2862
2863         if action_dict.get("stop"):
2864             action = "shutoff"
2865         db_new_tasks = []
2866         try:
2867             step = "lock the operation & do task creation"
2868             with self.write_lock:
2869                 extra_dict["params"] = {
2870                     "vim_vm_id": vim_vm_id,
2871                     "action": action,
2872                 }
2873                 task = self.rebuild_start_stop_task(
2874                     task_dict["vdu_id"],
2875                     task_dict["vnf_id"],
2876                     task_dict["vdu_index"],
2877                     action_id,
2878                     nsr_id,
2879                     task_index,
2880                     task_dict["target_vim"],
2881                     extra_dict,
2882                 )
2883                 db_new_tasks.append(task)
2884                 step = "upload Task to db"
2885                 self.upload_all_tasks(
2886                     db_new_tasks=db_new_tasks,
2887                     now=now,
2888                 )
2889                 self.logger.debug(
2890                     logging_text + "Exit. Created {} tasks".format(len(db_new_tasks))
2891                 )
2892                 return (
2893                     {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
2894                     action_id,
2895                     True,
2896                 )
2897         except Exception as e:
2898             if isinstance(e, (DbException, NsException)):
2899                 self.logger.error(
2900                     logging_text + "Exit Exception while '{}': {}".format(step, e)
2901                 )
2902             else:
2903                 e = traceback_format_exc()
2904                 self.logger.critical(
2905                     logging_text + "Exit Exception while '{}': {}".format(step, e),
2906                     exc_info=True,
2907                 )
2908             raise NsException(e)
2909
2910     def get_deploy(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
2911         nsrs = self.db.get_list("nsrs", {})
2912         return_data = []
2913
2914         for ns in nsrs:
2915             return_data.append({"_id": ns["_id"], "name": ns["name"]})
2916
2917         return return_data, None, True
2918
2919     def get_actions(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
2920         ro_tasks = self.db.get_list("ro_tasks", {"tasks.nsr_id": nsr_id})
2921         return_data = []
2922
2923         for ro_task in ro_tasks:
2924             for task in ro_task["tasks"]:
2925                 if task["action_id"] not in return_data:
2926                     return_data.append(task["action_id"])
2927
2928         return return_data, None, True
2929
2930     def migrate_task(
2931         self, vdu, vnf, vdu_index, action_id, nsr_id, task_index, extra_dict
2932     ):
2933         target_vim, vim_info = next(k_v for k_v in vdu["vim_info"].items())
2934         self._assign_vim(target_vim)
2935         target_record = "vnfrs:{}:vdur.{}".format(vnf["_id"], vdu_index)
2936         target_record_id = "vnfrs:{}:vdur.{}".format(vnf["_id"], vdu["id"])
2937         deployment_info = {
2938             "action_id": action_id,
2939             "nsr_id": nsr_id,
2940             "task_index": task_index,
2941         }
2942
2943         task = Ns._create_task(
2944             deployment_info=deployment_info,
2945             target_id=target_vim,
2946             item="migrate",
2947             action="EXEC",
2948             target_record=target_record,
2949             target_record_id=target_record_id,
2950             extra_dict=extra_dict,
2951         )
2952
2953         return task
2954
2955     def migrate(self, session, indata, version, nsr_id, *args, **kwargs):
2956         task_index = 0
2957         extra_dict = {}
2958         now = time()
2959         action_id = indata.get("action_id", str(uuid4()))
2960         step = ""
2961         logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id)
2962         self.logger.debug(logging_text + "Enter")
2963         try:
2964             vnf_instance_id = indata["vnfInstanceId"]
2965             step = "Getting vnfrs from db"
2966             db_vnfr = self.db.get_one("vnfrs", {"_id": vnf_instance_id})
2967             vdu = indata.get("vdu")
2968             migrateToHost = indata.get("migrateToHost")
2969             db_new_tasks = []
2970
2971             with self.write_lock:
2972                 if vdu is not None:
2973                     vdu_id = indata["vdu"]["vduId"]
2974                     vdu_count_index = indata["vdu"].get("vduCountIndex", 0)
2975                     for vdu_index, vdu in enumerate(db_vnfr["vdur"]):
2976                         if (
2977                             vdu["vdu-id-ref"] == vdu_id
2978                             and vdu["count-index"] == vdu_count_index
2979                         ):
2980                             extra_dict["params"] = {
2981                                 "vim_vm_id": vdu["vim-id"],
2982                                 "migrate_host": migrateToHost,
2983                                 "vdu_vim_info": vdu["vim_info"],
2984                             }
2985                             step = "Creating migration task for vdu:{}".format(vdu)
2986                             task = self.migrate_task(
2987                                 vdu,
2988                                 db_vnfr,
2989                                 vdu_index,
2990                                 action_id,
2991                                 nsr_id,
2992                                 task_index,
2993                                 extra_dict,
2994                             )
2995                             db_new_tasks.append(task)
2996                             task_index += 1
2997                             break
2998                 else:
2999                     for vdu_index, vdu in enumerate(db_vnfr["vdur"]):
3000                         extra_dict["params"] = {
3001                             "vim_vm_id": vdu["vim-id"],
3002                             "migrate_host": migrateToHost,
3003                             "vdu_vim_info": vdu["vim_info"],
3004                         }
3005                         step = "Creating migration task for vdu:{}".format(vdu)
3006                         task = self.migrate_task(
3007                             vdu,
3008                             db_vnfr,
3009                             vdu_index,
3010                             action_id,
3011                             nsr_id,
3012                             task_index,
3013                             extra_dict,
3014                         )
3015                         db_new_tasks.append(task)
3016                         task_index += 1
3017
3018                 self.upload_all_tasks(
3019                     db_new_tasks=db_new_tasks,
3020                     now=now,
3021                 )
3022
3023             self.logger.debug(
3024                 logging_text + "Exit. Created {} tasks".format(len(db_new_tasks))
3025             )
3026             return (
3027                 {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
3028                 action_id,
3029                 True,
3030             )
3031         except Exception as e:
3032             if isinstance(e, (DbException, NsException)):
3033                 self.logger.error(
3034                     logging_text + "Exit Exception while '{}': {}".format(step, e)
3035                 )
3036             else:
3037                 e = traceback_format_exc()
3038                 self.logger.critical(
3039                     logging_text + "Exit Exception while '{}': {}".format(step, e),
3040                     exc_info=True,
3041                 )
3042             raise NsException(e)
3043
3044     def verticalscale_task(
3045         self, vdu, vnf, vdu_index, action_id, nsr_id, task_index, extra_dict
3046     ):
3047         target_vim, vim_info = next(k_v for k_v in vdu["vim_info"].items())
3048         self._assign_vim(target_vim)
3049         target_record = "vnfrs:{}:vdur.{}".format(vnf["_id"], vdu_index)
3050         target_record_id = "vnfrs:{}:vdur.{}".format(vnf["_id"], vdu["id"])
3051         deployment_info = {
3052             "action_id": action_id,
3053             "nsr_id": nsr_id,
3054             "task_index": task_index,
3055         }
3056
3057         task = Ns._create_task(
3058             deployment_info=deployment_info,
3059             target_id=target_vim,
3060             item="verticalscale",
3061             action="EXEC",
3062             target_record=target_record,
3063             target_record_id=target_record_id,
3064             extra_dict=extra_dict,
3065         )
3066         return task
3067
3068     def verticalscale(self, session, indata, version, nsr_id, *args, **kwargs):
3069         task_index = 0
3070         extra_dict = {}
3071         now = time()
3072         action_id = indata.get("action_id", str(uuid4()))
3073         step = ""
3074         logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id)
3075         self.logger.debug(logging_text + "Enter")
3076         try:
3077             VnfFlavorData = indata.get("changeVnfFlavorData")
3078             vnf_instance_id = VnfFlavorData["vnfInstanceId"]
3079             step = "Getting vnfrs from db"
3080             db_vnfr = self.db.get_one("vnfrs", {"_id": vnf_instance_id})
3081             vduid = VnfFlavorData["additionalParams"]["vduid"]
3082             vduCountIndex = VnfFlavorData["additionalParams"]["vduCountIndex"]
3083             virtualMemory = VnfFlavorData["additionalParams"]["virtualMemory"]
3084             numVirtualCpu = VnfFlavorData["additionalParams"]["numVirtualCpu"]
3085             sizeOfStorage = VnfFlavorData["additionalParams"]["sizeOfStorage"]
3086             flavor_dict = {
3087                 "name": vduid + "-flv",
3088                 "ram": virtualMemory,
3089                 "vcpus": numVirtualCpu,
3090                 "disk": sizeOfStorage,
3091             }
3092             db_new_tasks = []
3093             step = "Creating Tasks for vertical scaling"
3094             with self.write_lock:
3095                 for vdu_index, vdu in enumerate(db_vnfr["vdur"]):
3096                     if (
3097                         vdu["vdu-id-ref"] == vduid
3098                         and vdu["count-index"] == vduCountIndex
3099                     ):
3100                         extra_dict["params"] = {
3101                             "vim_vm_id": vdu["vim-id"],
3102                             "flavor_dict": flavor_dict,
3103                         }
3104                         task = self.verticalscale_task(
3105                             vdu,
3106                             db_vnfr,
3107                             vdu_index,
3108                             action_id,
3109                             nsr_id,
3110                             task_index,
3111                             extra_dict,
3112                         )
3113                         db_new_tasks.append(task)
3114                         task_index += 1
3115                         break
3116                 self.upload_all_tasks(
3117                     db_new_tasks=db_new_tasks,
3118                     now=now,
3119                 )
3120             self.logger.debug(
3121                 logging_text + "Exit. Created {} tasks".format(len(db_new_tasks))
3122             )
3123             return (
3124                 {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
3125                 action_id,
3126                 True,
3127             )
3128         except Exception as e:
3129             if isinstance(e, (DbException, NsException)):
3130                 self.logger.error(
3131                     logging_text + "Exit Exception while '{}': {}".format(step, e)
3132                 )
3133             else:
3134                 e = traceback_format_exc()
3135                 self.logger.critical(
3136                     logging_text + "Exit Exception while '{}': {}".format(step, e),
3137                     exc_info=True,
3138                 )
3139             raise NsException(e)