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 import logging
  21 from random import choice as random_choice
  22 from threading import Lock
  23 from time import time
  24 from traceback import format_exc as traceback_format_exc
  25 from typing import Any, Dict, Tuple, Type
  26 from uuid import uuid4
  27
  28 from cryptography.hazmat.backends import default_backend as crypto_default_backend
  29 from cryptography.hazmat.primitives import serialization as crypto_serialization
  30 from cryptography.hazmat.primitives.asymmetric import rsa
  31 from jinja2 import (
  32     Environment,
  33     StrictUndefined,
  34     TemplateError,
  35     TemplateNotFound,
  36     UndefinedError,
  37 )
  38 from osm_common import (
  39     dbmemory,
  40     dbmongo,
  41     fslocal,
  42     fsmongo,
  43     msgkafka,
  44     msglocal,
  45     version as common_version,
  46 )
  47 from osm_common.dbbase import DbBase, DbException
  48 from osm_common.fsbase import FsBase, FsException
  49 from osm_common.msgbase import MsgException
  50 from osm_ng_ro.ns_thread import deep_get, NsWorker, NsWorkerException
  51 from osm_ng_ro.validation import deploy_schema, validate_input
  52 import yaml
  53
  54 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
  55 min_common_version = "0.1.16"
  56
  57
  58 class NsException(Exception):
  59     def __init__(self, message, http_code=HTTPStatus.BAD_REQUEST):
  60         self.http_code = http_code
  61         super(Exception, self).__init__(message)
  62
  63
  64 def get_process_id():
  65     """
  66     Obtain a unique ID for this process. If running from inside docker, it will get docker ID. If not it
  67     will provide a random one
  68     :return: Obtained ID
  69     """
  70     # Try getting docker id. If fails, get pid
  71     try:
  72         with open("/proc/self/cgroup", "r") as f:
  73             text_id_ = f.readline()
  74             _, _, text_id = text_id_.rpartition("/")
  75             text_id = text_id.replace("\n", "")[:12]
  76
  77             if text_id:
  78                 return text_id
  79     except Exception:
  80         pass
  81
  82     # Return a random id
  83     return "".join(random_choice("0123456789abcdef") for _ in range(12))
  84
  85
  86 def versiontuple(v):
  87     """utility for compare dot separate versions. Fills with zeros to proper number comparison"""
  88     filled = []
  89
  90     for point in v.split("."):
  91         filled.append(point.zfill(8))
  92
  93     return tuple(filled)
  94
  95
  96 class Ns(object):
  97     def __init__(self):
  98         self.db = None
  99         self.fs = None
 100         self.msg = None
 101         self.config = None
 102         # self.operations = None
 103         self.logger = None
 104         # ^ Getting logger inside method self.start because parent logger (ro) is not available yet.
 105         # If done now it will not be linked to parent not getting its handler and level
 106         self.map_topic = {}
 107         self.write_lock = None
 108         self.vims_assigned = {}
 109         self.next_worker = 0
 110         self.plugins = {}
 111         self.workers = []
 112         self.process_params_function_map = {
 113             "net": Ns._process_net_params,
 114             "image": Ns._process_image_params,
 115             "flavor": Ns._process_flavor_params,
 116             "vdu": Ns._process_vdu_params,
 117             "affinity-or-anti-affinity-group": Ns._process_affinity_group_params,
 118         }
 119         self.db_path_map = {
 120             "net": "vld",
 121             "image": "image",
 122             "flavor": "flavor",
 123             "vdu": "vdur",
 124             "affinity-or-anti-affinity-group": "affinity-or-anti-affinity-group",
 125         }
 126
 127     def init_db(self, target_version):
 128         pass
 129
 130     def start(self, config):
 131         """
 132         Connect to database, filesystem storage, and messaging
 133         :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
 134         :param config: Configuration of db, storage, etc
 135         :return: None
 136         """
 137         self.config = config
 138         self.config["process_id"] = get_process_id()  # used for HA identity
 139         self.logger = logging.getLogger("ro.ns")
 140
 141         # check right version of common
 142         if versiontuple(common_version) < versiontuple(min_common_version):
 143             raise NsException(
 144                 "Not compatible osm/common version '{}'. Needed '{}' or higher".format(
 145                     common_version, min_common_version
 146                 )
 147             )
 148
 149         try:
 150             if not self.db:
 151                 if config["database"]["driver"] == "mongo":
 152                     self.db = dbmongo.DbMongo()
 153                     self.db.db_connect(config["database"])
 154                 elif config["database"]["driver"] == "memory":
 155                     self.db = dbmemory.DbMemory()
 156                     self.db.db_connect(config["database"])
 157                 else:
 158                     raise NsException(
 159                         "Invalid configuration param '{}' at '[database]':'driver'".format(
 160                             config["database"]["driver"]
 161                         )
 162                     )
 163
 164             if not self.fs:
 165                 if config["storage"]["driver"] == "local":
 166                     self.fs = fslocal.FsLocal()
 167                     self.fs.fs_connect(config["storage"])
 168                 elif config["storage"]["driver"] == "mongo":
 169                     self.fs = fsmongo.FsMongo()
 170                     self.fs.fs_connect(config["storage"])
 171                 elif config["storage"]["driver"] is None:
 172                     pass
 173                 else:
 174                     raise NsException(
 175                         "Invalid configuration param '{}' at '[storage]':'driver'".format(
 176                             config["storage"]["driver"]
 177                         )
 178                     )
 179
 180             if not self.msg:
 181                 if config["message"]["driver"] == "local":
 182                     self.msg = msglocal.MsgLocal()
 183                     self.msg.connect(config["message"])
 184                 elif config["message"]["driver"] == "kafka":
 185                     self.msg = msgkafka.MsgKafka()
 186                     self.msg.connect(config["message"])
 187                 else:
 188                     raise NsException(
 189                         "Invalid configuration param '{}' at '[message]':'driver'".format(
 190                             config["message"]["driver"]
 191                         )
 192                     )
 193
 194             # TODO load workers to deal with exising database tasks
 195
 196             self.write_lock = Lock()
 197         except (DbException, FsException, MsgException) as e:
 198             raise NsException(str(e), http_code=e.http_code)
 199
 200     def get_assigned_vims(self):
 201         return list(self.vims_assigned.keys())
 202
 203     def stop(self):
 204         try:
 205             if self.db:
 206                 self.db.db_disconnect()
 207
 208             if self.fs:
 209                 self.fs.fs_disconnect()
 210
 211             if self.msg:
 212                 self.msg.disconnect()
 213
 214             self.write_lock = None
 215         except (DbException, FsException, MsgException) as e:
 216             raise NsException(str(e), http_code=e.http_code)
 217
 218         for worker in self.workers:
 219             worker.insert_task(("terminate",))
 220
 221     def _create_worker(self):
 222         """
 223         Look for a worker thread in idle status. If not found it creates one unless the number of threads reach the
 224         limit of 'server.ns_threads' configuration. If reached, it just assigns one existing thread
 225         return the index of the assigned worker thread. Worker threads are storead at self.workers
 226         """
 227         # Look for a thread in idle status
 228         worker_id = next(
 229             (
 230                 i
 231                 for i in range(len(self.workers))
 232                 if self.workers[i] and self.workers[i].idle
 233             ),
 234             None,
 235         )
 236
 237         if worker_id is not None:
 238             # unset idle status to avoid race conditions
 239             self.workers[worker_id].idle = False
 240         else:
 241             worker_id = len(self.workers)
 242
 243             if worker_id < self.config["global"]["server.ns_threads"]:
 244                 # create a new worker
 245                 self.workers.append(
 246                     NsWorker(worker_id, self.config, self.plugins, self.db)
 247                 )
 248                 self.workers[worker_id].start()
 249             else:
 250                 # reached maximum number of threads, assign VIM to an existing one
 251                 worker_id = self.next_worker
 252                 self.next_worker = (self.next_worker + 1) % self.config["global"][
 253                     "server.ns_threads"
 254                 ]
 255
 256         return worker_id
 257
 258     def assign_vim(self, target_id):
 259         with self.write_lock:
 260             return self._assign_vim(target_id)
 261
 262     def _assign_vim(self, target_id):
 263         if target_id not in self.vims_assigned:
 264             worker_id = self.vims_assigned[target_id] = self._create_worker()
 265             self.workers[worker_id].insert_task(("load_vim", target_id))
 266
 267     def reload_vim(self, target_id):
 268         # send reload_vim to the thread working with this VIM and inform all that a VIM has been changed,
 269         # this is because database VIM information is cached for threads working with SDN
 270         with self.write_lock:
 271             for worker in self.workers:
 272                 if worker and not worker.idle:
 273                     worker.insert_task(("reload_vim", target_id))
 274
 275     def unload_vim(self, target_id):
 276         with self.write_lock:
 277             return self._unload_vim(target_id)
 278
 279     def _unload_vim(self, target_id):
 280         if target_id in self.vims_assigned:
 281             worker_id = self.vims_assigned[target_id]
 282             self.workers[worker_id].insert_task(("unload_vim", target_id))
 283             del self.vims_assigned[target_id]
 284
 285     def check_vim(self, target_id):
 286         with self.write_lock:
 287             if target_id in self.vims_assigned:
 288                 worker_id = self.vims_assigned[target_id]
 289             else:
 290                 worker_id = self._create_worker()
 291
 292         worker = self.workers[worker_id]
 293         worker.insert_task(("check_vim", target_id))
 294
 295     def unload_unused_vims(self):
 296         with self.write_lock:
 297             vims_to_unload = []
 298
 299             for target_id in self.vims_assigned:
 300                 if not self.db.get_one(
 301                     "ro_tasks",
 302                     q_filter={
 303                         "target_id": target_id,
 304                         "tasks.status": ["SCHEDULED", "BUILD", "DONE", "FAILED"],
 305                     },
 306                     fail_on_empty=False,
 307                 ):
 308                     vims_to_unload.append(target_id)
 309
 310             for target_id in vims_to_unload:
 311                 self._unload_vim(target_id)
 312
 313     @staticmethod
 314     def _get_cloud_init(
 315         db: Type[DbBase],
 316         fs: Type[FsBase],
 317         location: str,
 318     ) -> str:
 319         """This method reads cloud init from a file.
 320
 321         Note: Not used as cloud init content is provided in the http body.
 322
 323         Args:
 324             db (Type[DbBase]): [description]
 325             fs (Type[FsBase]): [description]
 326             location (str): can be 'vnfr_id:file:file_name' or 'vnfr_id:vdu:vdu_idex'
 327
 328         Raises:
 329             NsException: [description]
 330             NsException: [description]
 331
 332         Returns:
 333             str: [description]
 334         """
 335         vnfd_id, _, other = location.partition(":")
 336         _type, _, name = other.partition(":")
 337         vnfd = db.get_one("vnfds", {"_id": vnfd_id})
 338
 339         if _type == "file":
 340             base_folder = vnfd["_admin"]["storage"]
 341             cloud_init_file = "{}/{}/cloud_init/{}".format(
 342                 base_folder["folder"], base_folder["pkg-dir"], name
 343             )
 344
 345             if not fs:
 346                 raise NsException(
 347                     "Cannot read file '{}'. Filesystem not loaded, change configuration at storage.driver".format(
 348                         cloud_init_file
 349                     )
 350                 )
 351
 352             with fs.file_open(cloud_init_file, "r") as ci_file:
 353                 cloud_init_content = ci_file.read()
 354         elif _type == "vdu":
 355             cloud_init_content = vnfd["vdu"][int(name)]["cloud-init"]
 356         else:
 357             raise NsException("Mismatch descriptor for cloud init: {}".format(location))
 358
 359         return cloud_init_content
 360
 361     @staticmethod
 362     def _parse_jinja2(
 363         cloud_init_content: str,
 364         params: Dict[str, Any],
 365         context: str,
 366     ) -> str:
 367         """Function that processes the cloud init to replace Jinja2 encoded parameters.
 368
 369         Args:
 370             cloud_init_content (str): [description]
 371             params (Dict[str, Any]): [description]
 372             context (str): [description]
 373
 374         Raises:
 375             NsException: [description]
 376             NsException: [description]
 377
 378         Returns:
 379             str: [description]
 380         """
 381         try:
 382             env = Environment(undefined=StrictUndefined)
 383             template = env.from_string(cloud_init_content)
 384
 385             return template.render(params or {})
 386         except UndefinedError as e:
 387             raise NsException(
 388                 "Variable '{}' defined at vnfd='{}' must be provided in the instantiation parameters"
 389                 "inside the 'additionalParamsForVnf' block".format(e, context)
 390             )
 391         except (TemplateError, TemplateNotFound) as e:
 392             raise NsException(
 393                 "Error parsing Jinja2 to cloud-init content at vnfd='{}': {}".format(
 394                     context, e
 395                 )
 396             )
 397
 398     def _create_db_ro_nsrs(self, nsr_id, now):
 399         try:
 400             key = rsa.generate_private_key(
 401                 backend=crypto_default_backend(), public_exponent=65537, key_size=2048
 402             )
 403             private_key = key.private_bytes(
 404                 crypto_serialization.Encoding.PEM,
 405                 crypto_serialization.PrivateFormat.PKCS8,
 406                 crypto_serialization.NoEncryption(),
 407             )
 408             public_key = key.public_key().public_bytes(
 409                 crypto_serialization.Encoding.OpenSSH,
 410                 crypto_serialization.PublicFormat.OpenSSH,
 411             )
 412             private_key = private_key.decode("utf8")
 413             # Change first line because Paramiko needs a explicit start with 'BEGIN RSA PRIVATE KEY'
 414             i = private_key.find("\n")
 415             private_key = "-----BEGIN RSA PRIVATE KEY-----" + private_key[i:]
 416             public_key = public_key.decode("utf8")
 417         except Exception as e:
 418             raise NsException("Cannot create ssh-keys: {}".format(e))
 419
 420         schema_version = "1.1"
 421         private_key_encrypted = self.db.encrypt(
 422             private_key, schema_version=schema_version, salt=nsr_id
 423         )
 424         db_content = {
 425             "_id": nsr_id,
 426             "_admin": {
 427                 "created": now,
 428                 "modified": now,
 429                 "schema_version": schema_version,
 430             },
 431             "public_key": public_key,
 432             "private_key": private_key_encrypted,
 433             "actions": [],
 434         }
 435         self.db.create("ro_nsrs", db_content)
 436
 437         return db_content
 438
 439     @staticmethod
 440     def _create_task(
 441         deployment_info: Dict[str, Any],
 442         target_id: str,
 443         item: str,
 444         action: str,
 445         target_record: str,
 446         target_record_id: str,
 447         extra_dict: Dict[str, Any] = None,
 448     ) -> Dict[str, Any]:
 449         """Function to create task dict from deployment information.
 450
 451         Args:
 452             deployment_info (Dict[str, Any]): [description]
 453             target_id (str): [description]
 454             item (str): [description]
 455             action (str): [description]
 456             target_record (str): [description]
 457             target_record_id (str): [description]
 458             extra_dict (Dict[str, Any], optional): [description]. Defaults to None.
 459
 460         Returns:
 461             Dict[str, Any]: [description]
 462         """
 463         task = {
 464             "target_id": target_id,  # it will be removed before pushing at database
 465             "action_id": deployment_info.get("action_id"),
 466             "nsr_id": deployment_info.get("nsr_id"),
 467             "task_id": f"{deployment_info.get('action_id')}:{deployment_info.get('task_index')}",
 468             "status": "SCHEDULED",
 469             "action": action,
 470             "item": item,
 471             "target_record": target_record,
 472             "target_record_id": target_record_id,
 473         }
 474
 475         if extra_dict:
 476             task.update(extra_dict)  # params, find_params, depends_on
 477
 478         deployment_info["task_index"] = deployment_info.get("task_index", 0) + 1
 479
 480         return task
 481
 482     @staticmethod
 483     def _create_ro_task(
 484         target_id: str,
 485         task: Dict[str, Any],
 486     ) -> Dict[str, Any]:
 487         """Function to create an RO task from task information.
 488
 489         Args:
 490             target_id (str): [description]
 491             task (Dict[str, Any]): [description]
 492
 493         Returns:
 494             Dict[str, Any]: [description]
 495         """
 496         now = time()
 497
 498         _id = task.get("task_id")
 499         db_ro_task = {
 500             "_id": _id,
 501             "locked_by": None,
 502             "locked_at": 0.0,
 503             "target_id": target_id,
 504             "vim_info": {
 505                 "created": False,
 506                 "created_items": None,
 507                 "vim_id": None,
 508                 "vim_name": None,
 509                 "vim_status": None,
 510                 "vim_details": None,
 511                 "refresh_at": None,
 512             },
 513             "modified_at": now,
 514             "created_at": now,
 515             "to_check_at": now,
 516             "tasks": [task],
 517         }
 518
 519         return db_ro_task
 520
 521     @staticmethod
 522     def _process_image_params(
 523         target_image: Dict[str, Any],
 524         indata: Dict[str, Any],
 525         vim_info: Dict[str, Any],
 526         target_record_id: str,
 527         **kwargs: Dict[str, Any],
 528     ) -> Dict[str, Any]:
 529         """Function to process VDU image parameters.
 530
 531         Args:
 532             target_image (Dict[str, Any]): [description]
 533             indata (Dict[str, Any]): [description]
 534             vim_info (Dict[str, Any]): [description]
 535             target_record_id (str): [description]
 536
 537         Returns:
 538             Dict[str, Any]: [description]
 539         """
 540         find_params = {}
 541
 542         if target_image.get("image"):
 543             find_params["filter_dict"] = {"name": target_image.get("image")}
 544
 545         if target_image.get("vim_image_id"):
 546             find_params["filter_dict"] = {"id": target_image.get("vim_image_id")}
 547
 548         if target_image.get("image_checksum"):
 549             find_params["filter_dict"] = {
 550                 "checksum": target_image.get("image_checksum")
 551             }
 552
 553         return {"find_params": find_params}
 554
 555     @staticmethod
 556     def _get_resource_allocation_params(
 557         quota_descriptor: Dict[str, Any],
 558     ) -> Dict[str, Any]:
 559         """Read the quota_descriptor from vnfd and fetch the resource allocation properties from the
 560         descriptor object.
 561
 562         Args:
 563             quota_descriptor (Dict[str, Any]): cpu/mem/vif/disk-io quota descriptor
 564
 565         Returns:
 566             Dict[str, Any]: quota params for limit, reserve, shares from the descriptor object
 567         """
 568         quota = {}
 569
 570         if quota_descriptor.get("limit"):
 571             quota["limit"] = int(quota_descriptor["limit"])
 572
 573         if quota_descriptor.get("reserve"):
 574             quota["reserve"] = int(quota_descriptor["reserve"])
 575
 576         if quota_descriptor.get("shares"):
 577             quota["shares"] = int(quota_descriptor["shares"])
 578
 579         return quota
 580
 581     @staticmethod
 582     def _process_guest_epa_quota_params(
 583         guest_epa_quota: Dict[str, Any],
 584         epa_vcpu_set: bool,
 585     ) -> Dict[str, Any]:
 586         """Function to extract the guest epa quota parameters.
 587
 588         Args:
 589             guest_epa_quota (Dict[str, Any]): [description]
 590             epa_vcpu_set (bool): [description]
 591
 592         Returns:
 593             Dict[str, Any]: [description]
 594         """
 595         result = {}
 596
 597         if guest_epa_quota.get("cpu-quota") and not epa_vcpu_set:
 598             cpuquota = Ns._get_resource_allocation_params(
 599                 guest_epa_quota.get("cpu-quota")
 600             )
 601
 602             if cpuquota:
 603                 result["cpu-quota"] = cpuquota
 604
 605         if guest_epa_quota.get("mem-quota"):
 606             vduquota = Ns._get_resource_allocation_params(
 607                 guest_epa_quota.get("mem-quota")
 608             )
 609
 610             if vduquota:
 611                 result["mem-quota"] = vduquota
 612
 613         if guest_epa_quota.get("disk-io-quota"):
 614             diskioquota = Ns._get_resource_allocation_params(
 615                 guest_epa_quota.get("disk-io-quota")
 616             )
 617
 618             if diskioquota:
 619                 result["disk-io-quota"] = diskioquota
 620
 621         if guest_epa_quota.get("vif-quota"):
 622             vifquota = Ns._get_resource_allocation_params(
 623                 guest_epa_quota.get("vif-quota")
 624             )
 625
 626             if vifquota:
 627                 result["vif-quota"] = vifquota
 628
 629         return result
 630
 631     @staticmethod
 632     def _process_guest_epa_numa_params(
 633         guest_epa_quota: Dict[str, Any],
 634     ) -> Tuple[Dict[str, Any], bool]:
 635         """[summary]
 636
 637         Args:
 638             guest_epa_quota (Dict[str, Any]): [description]
 639
 640         Returns:
 641             Tuple[Dict[str, Any], bool]: [description]
 642         """
 643         numa = {}
 644         epa_vcpu_set = False
 645
 646         if guest_epa_quota.get("numa-node-policy"):
 647             numa_node_policy = guest_epa_quota.get("numa-node-policy")
 648
 649             if numa_node_policy.get("node"):
 650                 numa_node = numa_node_policy["node"][0]
 651
 652                 if numa_node.get("num-cores"):
 653                     numa["cores"] = numa_node["num-cores"]
 654                     epa_vcpu_set = True
 655
 656                 paired_threads = numa_node.get("paired-threads", {})
 657                 if paired_threads.get("num-paired-threads"):
 658                     numa["paired-threads"] = int(
 659                         numa_node["paired-threads"]["num-paired-threads"]
 660                     )
 661                     epa_vcpu_set = True
 662
 663                 if paired_threads.get("paired-thread-ids"):
 664                     numa["paired-threads-id"] = []
 665
 666                     for pair in paired_threads["paired-thread-ids"]:
 667                         numa["paired-threads-id"].append(
 668                             (
 669                                 str(pair["thread-a"]),
 670                                 str(pair["thread-b"]),
 671                             )
 672                         )
 673
 674                 if numa_node.get("num-threads"):
 675                     numa["threads"] = int(numa_node["num-threads"])
 676                     epa_vcpu_set = True
 677
 678                 if numa_node.get("memory-mb"):
 679                     numa["memory"] = max(int(int(numa_node["memory-mb"]) / 1024), 1)
 680
 681         return numa, epa_vcpu_set
 682
 683     @staticmethod
 684     def _process_guest_epa_cpu_pinning_params(
 685         guest_epa_quota: Dict[str, Any],
 686         vcpu_count: int,
 687         epa_vcpu_set: bool,
 688     ) -> Tuple[Dict[str, Any], bool]:
 689         """[summary]
 690
 691         Args:
 692             guest_epa_quota (Dict[str, Any]): [description]
 693             vcpu_count (int): [description]
 694             epa_vcpu_set (bool): [description]
 695
 696         Returns:
 697             Tuple[Dict[str, Any], bool]: [description]
 698         """
 699         numa = {}
 700         local_epa_vcpu_set = epa_vcpu_set
 701
 702         if (
 703             guest_epa_quota.get("cpu-pinning-policy") == "DEDICATED"
 704             and not epa_vcpu_set
 705         ):
 706             numa[
 707                 "cores"
 708                 if guest_epa_quota.get("cpu-thread-pinning-policy") != "PREFER"
 709                 else "threads"
 710             ] = max(vcpu_count, 1)
 711             local_epa_vcpu_set = True
 712
 713         return numa, local_epa_vcpu_set
 714
 715     @staticmethod
 716     def _process_epa_params(
 717         target_flavor: Dict[str, Any],
 718     ) -> Dict[str, Any]:
 719         """[summary]
 720
 721         Args:
 722             target_flavor (Dict[str, Any]): [description]
 723
 724         Returns:
 725             Dict[str, Any]: [description]
 726         """
 727         extended = {}
 728         numa = {}
 729
 730         if target_flavor.get("guest-epa"):
 731             guest_epa = target_flavor["guest-epa"]
 732
 733             numa, epa_vcpu_set = Ns._process_guest_epa_numa_params(
 734                 guest_epa_quota=guest_epa
 735             )
 736
 737             if guest_epa.get("mempage-size"):
 738                 extended["mempage-size"] = guest_epa.get("mempage-size")
 739
 740             tmp_numa, epa_vcpu_set = Ns._process_guest_epa_cpu_pinning_params(
 741                 guest_epa_quota=guest_epa,
 742                 vcpu_count=int(target_flavor.get("vcpu-count", 1)),
 743                 epa_vcpu_set=epa_vcpu_set,
 744             )
 745             numa.update(tmp_numa)
 746
 747             extended.update(
 748                 Ns._process_guest_epa_quota_params(
 749                     guest_epa_quota=guest_epa,
 750                     epa_vcpu_set=epa_vcpu_set,
 751                 )
 752             )
 753
 754         if numa:
 755             extended["numas"] = [numa]
 756
 757         return extended
 758
 759     @staticmethod
 760     def _process_flavor_params(
 761         target_flavor: Dict[str, Any],
 762         indata: Dict[str, Any],
 763         vim_info: Dict[str, Any],
 764         target_record_id: str,
 765         **kwargs: Dict[str, Any],
 766     ) -> Dict[str, Any]:
 767         """[summary]
 768
 769         Args:
 770             target_flavor (Dict[str, Any]): [description]
 771             indata (Dict[str, Any]): [description]
 772             vim_info (Dict[str, Any]): [description]
 773             target_record_id (str): [description]
 774
 775         Returns:
 776             Dict[str, Any]: [description]
 777         """
 778         flavor_data = {
 779             "disk": int(target_flavor["storage-gb"]),
 780             "ram": int(target_flavor["memory-mb"]),
 781             "vcpus": int(target_flavor["vcpu-count"]),
 782         }
 783
 784         target_vdur = {}
 785         for vnf in indata.get("vnf", []):
 786             for vdur in vnf.get("vdur", []):
 787                 if vdur.get("ns-flavor-id") == target_flavor["id"]:
 788                     target_vdur = vdur
 789
 790         for storage in target_vdur.get("virtual-storages", []):
 791             if (
 792                 storage.get("type-of-storage")
 793                 == "etsi-nfv-descriptors:ephemeral-storage"
 794             ):
 795                 flavor_data["ephemeral"] = int(storage.get("size-of-storage", 0))
 796             elif storage.get("type-of-storage") == "etsi-nfv-descriptors:swap-storage":
 797                 flavor_data["swap"] = int(storage.get("size-of-storage", 0))
 798
 799         extended = Ns._process_epa_params(target_flavor)
 800         if extended:
 801             flavor_data["extended"] = extended
 802
 803         extra_dict = {"find_params": {"flavor_data": flavor_data}}
 804         flavor_data_name = flavor_data.copy()
 805         flavor_data_name["name"] = target_flavor["name"]
 806         extra_dict["params"] = {"flavor_data": flavor_data_name}
 807
 808         return extra_dict
 809
 810     @staticmethod
 811     def _ip_profile_to_ro(
 812         ip_profile: Dict[str, Any],
 813     ) -> Dict[str, Any]:
 814         """[summary]
 815
 816         Args:
 817             ip_profile (Dict[str, Any]): [description]
 818
 819         Returns:
 820             Dict[str, Any]: [description]
 821         """
 822         if not ip_profile:
 823             return None
 824
 825         ro_ip_profile = {
 826             "ip_version": "IPv4"
 827             if "v4" in ip_profile.get("ip-version", "ipv4")
 828             else "IPv6",
 829             "subnet_address": ip_profile.get("subnet-address"),
 830             "gateway_address": ip_profile.get("gateway-address"),
 831             "dhcp_enabled": ip_profile.get("dhcp-params", {}).get("enabled", False),
 832             "dhcp_start_address": ip_profile.get("dhcp-params", {}).get(
 833                 "start-address", None
 834             ),
 835             "dhcp_count": ip_profile.get("dhcp-params", {}).get("count", None),
 836         }
 837
 838         if ip_profile.get("dns-server"):
 839             ro_ip_profile["dns_address"] = ";".join(
 840                 [v["address"] for v in ip_profile["dns-server"] if v.get("address")]
 841             )
 842
 843         if ip_profile.get("security-group"):
 844             ro_ip_profile["security_group"] = ip_profile["security-group"]
 845
 846         return ro_ip_profile
 847
 848     @staticmethod
 849     def _process_net_params(
 850         target_vld: Dict[str, Any],
 851         indata: Dict[str, Any],
 852         vim_info: Dict[str, Any],
 853         target_record_id: str,
 854         **kwargs: Dict[str, Any],
 855     ) -> Dict[str, Any]:
 856         """Function to process network parameters.
 857
 858         Args:
 859             target_vld (Dict[str, Any]): [description]
 860             indata (Dict[str, Any]): [description]
 861             vim_info (Dict[str, Any]): [description]
 862             target_record_id (str): [description]
 863
 864         Returns:
 865             Dict[str, Any]: [description]
 866         """
 867         extra_dict = {}
 868
 869         if vim_info.get("sdn"):
 870             # vnf_preffix = "vnfrs:{}".format(vnfr_id)
 871             # ns_preffix = "nsrs:{}".format(nsr_id)
 872             # remove the ending ".sdn
 873             vld_target_record_id, _, _ = target_record_id.rpartition(".")
 874             extra_dict["params"] = {
 875                 k: vim_info[k]
 876                 for k in ("sdn-ports", "target_vim", "vlds", "type")
 877                 if vim_info.get(k)
 878             }
 879
 880             # TODO needed to add target_id in the dependency.
 881             if vim_info.get("target_vim"):
 882                 extra_dict["depends_on"] = [
 883                     f"{vim_info.get('target_vim')} {vld_target_record_id}"
 884                 ]
 885
 886             return extra_dict
 887
 888         if vim_info.get("vim_network_name"):
 889             extra_dict["find_params"] = {
 890                 "filter_dict": {
 891                     "name": vim_info.get("vim_network_name"),
 892                 },
 893             }
 894         elif vim_info.get("vim_network_id"):
 895             extra_dict["find_params"] = {
 896                 "filter_dict": {
 897                     "id": vim_info.get("vim_network_id"),
 898                 },
 899             }
 900         elif target_vld.get("mgmt-network"):
 901             extra_dict["find_params"] = {
 902                 "mgmt": True,
 903                 "name": target_vld["id"],
 904             }
 905         else:
 906             # create
 907             extra_dict["params"] = {
 908                 "net_name": (
 909                     f"{indata.get('name')[:16]}-{target_vld.get('name', target_vld.get('id'))[:16]}"
 910                 ),
 911                 "ip_profile": Ns._ip_profile_to_ro(vim_info.get("ip_profile")),
 912                 "provider_network_profile": vim_info.get("provider_network"),
 913             }
 914
 915             if not target_vld.get("underlay"):
 916                 extra_dict["params"]["net_type"] = "bridge"
 917             else:
 918                 extra_dict["params"]["net_type"] = (
 919                     "ptp" if target_vld.get("type") == "ELINE" else "data"
 920                 )
 921
 922         return extra_dict
 923
 924     @staticmethod
 925     def _process_vdu_params(
 926         target_vdu: Dict[str, Any],
 927         indata: Dict[str, Any],
 928         vim_info: Dict[str, Any],
 929         target_record_id: str,
 930         **kwargs: Dict[str, Any],
 931     ) -> Dict[str, Any]:
 932         """Function to process VDU parameters.
 933
 934         Args:
 935             target_vdu (Dict[str, Any]): [description]
 936             indata (Dict[str, Any]): [description]
 937             vim_info (Dict[str, Any]): [description]
 938             target_record_id (str): [description]
 939
 940         Returns:
 941             Dict[str, Any]: [description]
 942         """
 943         vnfr_id = kwargs.get("vnfr_id")
 944         nsr_id = kwargs.get("nsr_id")
 945         vnfr = kwargs.get("vnfr")
 946         vdu2cloud_init = kwargs.get("vdu2cloud_init")
 947         tasks_by_target_record_id = kwargs.get("tasks_by_target_record_id")
 948         logger = kwargs.get("logger")
 949         db = kwargs.get("db")
 950         fs = kwargs.get("fs")
 951         ro_nsr_public_key = kwargs.get("ro_nsr_public_key")
 952
 953         vnf_preffix = "vnfrs:{}".format(vnfr_id)
 954         ns_preffix = "nsrs:{}".format(nsr_id)
 955         image_text = ns_preffix + ":image." + target_vdu["ns-image-id"]
 956         flavor_text = ns_preffix + ":flavor." + target_vdu["ns-flavor-id"]
 957         extra_dict = {"depends_on": [image_text, flavor_text]}
 958         net_list = []
 959
 960         for iface_index, interface in enumerate(target_vdu["interfaces"]):
 961             if interface.get("ns-vld-id"):
 962                 net_text = ns_preffix + ":vld." + interface["ns-vld-id"]
 963             elif interface.get("vnf-vld-id"):
 964                 net_text = vnf_preffix + ":vld." + interface["vnf-vld-id"]
 965             else:
 966                 logger.error(
 967                     "Interface {} from vdu {} not connected to any vld".format(
 968                         iface_index, target_vdu["vdu-name"]
 969                     )
 970                 )
 971
 972                 continue  # interface not connected to any vld
 973
 974             extra_dict["depends_on"].append(net_text)
 975
 976             if "port-security-enabled" in interface:
 977                 interface["port_security"] = interface.pop("port-security-enabled")
 978
 979             if "port-security-disable-strategy" in interface:
 980                 interface["port_security_disable_strategy"] = interface.pop(
 981                     "port-security-disable-strategy"
 982                 )
 983
 984             net_item = {
 985                 x: v
 986                 for x, v in interface.items()
 987                 if x
 988                 in (
 989                     "name",
 990                     "vpci",
 991                     "port_security",
 992                     "port_security_disable_strategy",
 993                     "floating_ip",
 994                 )
 995             }
 996             net_item["net_id"] = "TASK-" + net_text
 997             net_item["type"] = "virtual"
 998
 999             # TODO mac_address: used for  SR-IOV ifaces #TODO for other types
1000             # TODO floating_ip: True/False (or it can be None)
1001             if interface.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
1002                 # mark the net create task as type data
1003                 if deep_get(
1004                     tasks_by_target_record_id,
1005                     net_text,
1006                     "extra_dict",
1007                     "params",
1008                     "net_type",
1009                 ):
1010                     tasks_by_target_record_id[net_text]["extra_dict"]["params"][
1011                         "net_type"
1012                     ] = "data"
1013
1014                 net_item["use"] = "data"
1015                 net_item["model"] = interface["type"]
1016                 net_item["type"] = interface["type"]
1017             elif (
1018                 interface.get("type") == "OM-MGMT"
1019                 or interface.get("mgmt-interface")
1020                 or interface.get("mgmt-vnf")
1021             ):
1022                 net_item["use"] = "mgmt"
1023             else:
1024                 # if interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"):
1025                 net_item["use"] = "bridge"
1026                 net_item["model"] = interface.get("type")
1027
1028             if interface.get("ip-address"):
1029                 net_item["ip_address"] = interface["ip-address"]
1030
1031             if interface.get("mac-address"):
1032                 net_item["mac_address"] = interface["mac-address"]
1033
1034             net_list.append(net_item)
1035
1036             if interface.get("mgmt-vnf"):
1037                 extra_dict["mgmt_vnf_interface"] = iface_index
1038             elif interface.get("mgmt-interface"):
1039                 extra_dict["mgmt_vdu_interface"] = iface_index
1040
1041         # cloud config
1042         cloud_config = {}
1043
1044         if target_vdu.get("cloud-init"):
1045             if target_vdu["cloud-init"] not in vdu2cloud_init:
1046                 vdu2cloud_init[target_vdu["cloud-init"]] = Ns._get_cloud_init(
1047                     db=db,
1048                     fs=fs,
1049                     location=target_vdu["cloud-init"],
1050                 )
1051
1052             cloud_content_ = vdu2cloud_init[target_vdu["cloud-init"]]
1053             cloud_config["user-data"] = Ns._parse_jinja2(
1054                 cloud_init_content=cloud_content_,
1055                 params=target_vdu.get("additionalParams"),
1056                 context=target_vdu["cloud-init"],
1057             )
1058
1059         if target_vdu.get("boot-data-drive"):
1060             cloud_config["boot-data-drive"] = target_vdu.get("boot-data-drive")
1061
1062         ssh_keys = []
1063
1064         if target_vdu.get("ssh-keys"):
1065             ssh_keys += target_vdu.get("ssh-keys")
1066
1067         if target_vdu.get("ssh-access-required"):
1068             ssh_keys.append(ro_nsr_public_key)
1069
1070         if ssh_keys:
1071             cloud_config["key-pairs"] = ssh_keys
1072
1073         persistent_root_disk = {}
1074         disk_list = []
1075         vnfd_id = vnfr["vnfd-id"]
1076         vnfd = db.get_one("vnfds", {"_id": vnfd_id})
1077         for vdu in vnfd.get("vdu", ()):
1078             if vdu["name"] == target_vdu["vdu-name"]:
1079                 for vsd in vnfd.get("virtual-storage-desc", ()):
1080                     if vsd.get("id") == vdu.get("virtual-storage-desc", [[]])[0]:
1081                         root_disk = vsd
1082                         if root_disk.get(
1083                             "type-of-storage"
1084                         ) == "persistent-storage:persistent-storage" and root_disk.get(
1085                             "size-of-storage"
1086                         ):
1087                             persistent_root_disk[vsd["id"]] = {
1088                                 "image_id": vdu.get("sw-image-desc"),
1089                                 "size": root_disk["size-of-storage"],
1090                             }
1091                             disk_list.append(persistent_root_disk[vsd["id"]])
1092
1093         if target_vdu.get("virtual-storages"):
1094             for disk in target_vdu["virtual-storages"]:
1095                 if (
1096                     disk.get("type-of-storage")
1097                     == "persistent-storage:persistent-storage"
1098                     and disk["id"] not in persistent_root_disk.keys()
1099                 ):
1100                     disk_list.append({"size": disk["size-of-storage"]})
1101
1102         affinity_group_list = []
1103
1104         if target_vdu.get("affinity-or-anti-affinity-group-id"):
1105             affinity_group = {}
1106             for affinity_group_id in target_vdu["affinity-or-anti-affinity-group-id"]:
1107                 affinity_group_text = (
1108                     ns_preffix + ":affinity-or-anti-affinity-group." + affinity_group_id
1109                 )
1110
1111                 extra_dict["depends_on"].append(affinity_group_text)
1112                 affinity_group["affinity_group_id"] = "TASK-" + affinity_group_text
1113                 affinity_group_list.append(affinity_group)
1114
1115         extra_dict["params"] = {
1116             "name": "{}-{}-{}-{}".format(
1117                 indata["name"][:16],
1118                 vnfr["member-vnf-index-ref"][:16],
1119                 target_vdu["vdu-name"][:32],
1120                 target_vdu.get("count-index") or 0,
1121             ),
1122             "description": target_vdu["vdu-name"],
1123             "start": True,
1124             "image_id": "TASK-" + image_text,
1125             "flavor_id": "TASK-" + flavor_text,
1126             "affinity_group_list": affinity_group_list,
1127             "net_list": net_list,
1128             "cloud_config": cloud_config or None,
1129             "disk_list": disk_list,
1130             "availability_zone_index": None,  # TODO
1131             "availability_zone_list": None,  # TODO
1132         }
1133
1134         return extra_dict
1135
1136     @staticmethod
1137     def _process_affinity_group_params(
1138         target_affinity_group: Dict[str, Any],
1139         indata: Dict[str, Any],
1140         vim_info: Dict[str, Any],
1141         target_record_id: str,
1142         **kwargs: Dict[str, Any],
1143     ) -> Dict[str, Any]:
1144         """Get affinity or anti-affinity group parameters.
1145
1146         Args:
1147             target_affinity_group (Dict[str, Any]): [description]
1148             indata (Dict[str, Any]): [description]
1149             vim_info (Dict[str, Any]): [description]
1150             target_record_id (str): [description]
1151
1152         Returns:
1153             Dict[str, Any]: [description]
1154         """
1155
1156         extra_dict = {}
1157         affinity_group_data = {
1158             "name": target_affinity_group["name"],
1159             "type": target_affinity_group["type"],
1160             "scope": target_affinity_group["scope"],
1161         }
1162
1163         if target_affinity_group.get("vim-affinity-group-id"):
1164             affinity_group_data["vim-affinity-group-id"] = target_affinity_group[
1165                 "vim-affinity-group-id"
1166             ]
1167
1168         extra_dict["params"] = {
1169             "affinity_group_data": affinity_group_data,
1170         }
1171
1172         return extra_dict
1173
1174     @staticmethod
1175     def _process_recreate_vdu_params(
1176         existing_vdu: Dict[str, Any],
1177         db_nsr: Dict[str, Any],
1178         vim_info: Dict[str, Any],
1179         target_record_id: str,
1180         target_id: str,
1181         **kwargs: Dict[str, Any],
1182     ) -> Dict[str, Any]:
1183         """Function to process VDU parameters to recreate.
1184
1185         Args:
1186             existing_vdu (Dict[str, Any]): [description]
1187             db_nsr (Dict[str, Any]): [description]
1188             vim_info (Dict[str, Any]): [description]
1189             target_record_id (str): [description]
1190             target_id (str): [description]
1191
1192         Returns:
1193             Dict[str, Any]: [description]
1194         """
1195         vnfr = kwargs.get("vnfr")
1196         vdu2cloud_init = kwargs.get("vdu2cloud_init")
1197         # logger = kwargs.get("logger")
1198         db = kwargs.get("db")
1199         fs = kwargs.get("fs")
1200         ro_nsr_public_key = kwargs.get("ro_nsr_public_key")
1201
1202         extra_dict = {}
1203         net_list = []
1204
1205         vim_details = {}
1206         vim_details_text = existing_vdu["vim_info"][target_id].get("vim_details", None)
1207         if vim_details_text:
1208             vim_details = yaml.safe_load(f"{vim_details_text}")
1209
1210         for iface_index, interface in enumerate(existing_vdu["interfaces"]):
1211
1212             if "port-security-enabled" in interface:
1213                 interface["port_security"] = interface.pop("port-security-enabled")
1214
1215             if "port-security-disable-strategy" in interface:
1216                 interface["port_security_disable_strategy"] = interface.pop(
1217                     "port-security-disable-strategy"
1218                 )
1219
1220             net_item = {
1221                 x: v
1222                 for x, v in interface.items()
1223                 if x
1224                 in (
1225                     "name",
1226                     "vpci",
1227                     "port_security",
1228                     "port_security_disable_strategy",
1229                     "floating_ip",
1230                 )
1231             }
1232             existing_ifaces = existing_vdu["vim_info"][target_id].get("interfaces", [])
1233             net_id = next(
1234                 (
1235                     i["vim_net_id"]
1236                     for i in existing_ifaces
1237                     if i["ip_address"] == interface["ip-address"]
1238                 ),
1239                 None,
1240             )
1241
1242             net_item["net_id"] = net_id
1243             net_item["type"] = "virtual"
1244
1245             # TODO mac_address: used for  SR-IOV ifaces #TODO for other types
1246             # TODO floating_ip: True/False (or it can be None)
1247             if interface.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
1248                 net_item["use"] = "data"
1249                 net_item["model"] = interface["type"]
1250                 net_item["type"] = interface["type"]
1251             elif (
1252                 interface.get("type") == "OM-MGMT"
1253                 or interface.get("mgmt-interface")
1254                 or interface.get("mgmt-vnf")
1255             ):
1256                 net_item["use"] = "mgmt"
1257             else:
1258                 # if interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"):
1259                 net_item["use"] = "bridge"
1260                 net_item["model"] = interface.get("type")
1261
1262             if interface.get("ip-address"):
1263                 net_item["ip_address"] = interface["ip-address"]
1264
1265             if interface.get("mac-address"):
1266                 net_item["mac_address"] = interface["mac-address"]
1267
1268             net_list.append(net_item)
1269
1270             if interface.get("mgmt-vnf"):
1271                 extra_dict["mgmt_vnf_interface"] = iface_index
1272             elif interface.get("mgmt-interface"):
1273                 extra_dict["mgmt_vdu_interface"] = iface_index
1274
1275         # cloud config
1276         cloud_config = {}
1277
1278         if existing_vdu.get("cloud-init"):
1279             if existing_vdu["cloud-init"] not in vdu2cloud_init:
1280                 vdu2cloud_init[existing_vdu["cloud-init"]] = Ns._get_cloud_init(
1281                     db=db,
1282                     fs=fs,
1283                     location=existing_vdu["cloud-init"],
1284                 )
1285
1286             cloud_content_ = vdu2cloud_init[existing_vdu["cloud-init"]]
1287             cloud_config["user-data"] = Ns._parse_jinja2(
1288                 cloud_init_content=cloud_content_,
1289                 params=existing_vdu.get("additionalParams"),
1290                 context=existing_vdu["cloud-init"],
1291             )
1292
1293         if existing_vdu.get("boot-data-drive"):
1294             cloud_config["boot-data-drive"] = existing_vdu.get("boot-data-drive")
1295
1296         ssh_keys = []
1297
1298         if existing_vdu.get("ssh-keys"):
1299             ssh_keys += existing_vdu.get("ssh-keys")
1300
1301         if existing_vdu.get("ssh-access-required"):
1302             ssh_keys.append(ro_nsr_public_key)
1303
1304         if ssh_keys:
1305             cloud_config["key-pairs"] = ssh_keys
1306
1307         disk_list = []
1308         for vol_id in vim_details.get("os-extended-volumes:volumes_attached", []):
1309             disk_list.append({"vim_id": vol_id["id"]})
1310
1311         affinity_group_list = []
1312
1313         if existing_vdu.get("affinity-or-anti-affinity-group-id"):
1314             affinity_group = {}
1315             for affinity_group_id in existing_vdu["affinity-or-anti-affinity-group-id"]:
1316                 for group in db_nsr.get("affinity-or-anti-affinity-group"):
1317                     if (
1318                         group["id"] == affinity_group_id
1319                         and group["vim_info"][target_id].get("vim_id", None) is not None
1320                     ):
1321                         affinity_group["affinity_group_id"] = group["vim_info"][
1322                             target_id
1323                         ].get("vim_id", None)
1324                         affinity_group_list.append(affinity_group)
1325
1326         extra_dict["params"] = {
1327             "name": "{}-{}-{}-{}".format(
1328                 db_nsr["name"][:16],
1329                 vnfr["member-vnf-index-ref"][:16],
1330                 existing_vdu["vdu-name"][:32],
1331                 existing_vdu.get("count-index") or 0,
1332             ),
1333             "description": existing_vdu["vdu-name"],
1334             "start": True,
1335             "image_id": vim_details["image"]["id"],
1336             "flavor_id": vim_details["flavor"]["id"],
1337             "affinity_group_list": affinity_group_list,
1338             "net_list": net_list,
1339             "cloud_config": cloud_config or None,
1340             "disk_list": disk_list,
1341             "availability_zone_index": None,  # TODO
1342             "availability_zone_list": None,  # TODO
1343         }
1344
1345         return extra_dict
1346
1347     def calculate_diff_items(
1348         self,
1349         indata,
1350         db_nsr,
1351         db_ro_nsr,
1352         db_nsr_update,
1353         item,
1354         tasks_by_target_record_id,
1355         action_id,
1356         nsr_id,
1357         task_index,
1358         vnfr_id=None,
1359         vnfr=None,
1360     ):
1361         """Function that returns the incremental changes (creation, deletion)
1362         related to a specific item `item` to be done. This function should be
1363         called for NS instantiation, NS termination, NS update to add a new VNF
1364         or a new VLD, remove a VNF or VLD, etc.
1365         Item can be `net`, `flavor`, `image` or `vdu`.
1366         It takes a list of target items from indata (which came from the REST API)
1367         and compares with the existing items from db_ro_nsr, identifying the
1368         incremental changes to be done. During the comparison, it calls the method
1369         `process_params` (which was passed as parameter, and is particular for each
1370         `item`)
1371
1372         Args:
1373             indata (Dict[str, Any]): deployment info
1374             db_nsr: NSR record from DB
1375             db_ro_nsr (Dict[str, Any]): record from "ro_nsrs"
1376             db_nsr_update (Dict[str, Any]): NSR info to update in DB
1377             item (str): element to process (net, vdu...)
1378             tasks_by_target_record_id (Dict[str, Any]):
1379                 [<target_record_id>, <task>]
1380             action_id (str): action id
1381             nsr_id (str): NSR id
1382             task_index (number): task index to add to task name
1383             vnfr_id (str): VNFR id
1384             vnfr (Dict[str, Any]): VNFR info
1385
1386         Returns:
1387             List: list with the incremental changes (deletes, creates) for each item
1388             number: current task index
1389         """
1390
1391         diff_items = []
1392         db_path = ""
1393         db_record = ""
1394         target_list = []
1395         existing_list = []
1396         process_params = None
1397         vdu2cloud_init = indata.get("cloud_init_content") or {}
1398         ro_nsr_public_key = db_ro_nsr["public_key"]
1399
1400         # According to the type of item, the path, the target_list,
1401         # the existing_list and the method to process params are set
1402         db_path = self.db_path_map[item]
1403         process_params = self.process_params_function_map[item]
1404         if item in ("net", "vdu"):
1405             # This case is specific for the NS VLD (not applied to VDU)
1406             if vnfr is None:
1407                 db_record = "nsrs:{}:{}".format(nsr_id, db_path)
1408                 target_list = indata.get("ns", []).get(db_path, [])
1409                 existing_list = db_nsr.get(db_path, [])
1410             # This case is common for VNF VLDs and VNF VDUs
1411             else:
1412                 db_record = "vnfrs:{}:{}".format(vnfr_id, db_path)
1413                 target_vnf = next(
1414                     (vnf for vnf in indata.get("vnf", ()) if vnf["_id"] == vnfr_id),
1415                     None,
1416                 )
1417                 target_list = target_vnf.get(db_path, []) if target_vnf else []
1418                 existing_list = vnfr.get(db_path, [])
1419         elif item in ("image", "flavor", "affinity-or-anti-affinity-group"):
1420             db_record = "nsrs:{}:{}".format(nsr_id, db_path)
1421             target_list = indata.get(item, [])
1422             existing_list = db_nsr.get(item, [])
1423         else:
1424             raise NsException("Item not supported: {}", item)
1425
1426         # ensure all the target_list elements has an "id". If not assign the index as id
1427         if target_list is None:
1428             target_list = []
1429         for target_index, tl in enumerate(target_list):
1430             if tl and not tl.get("id"):
1431                 tl["id"] = str(target_index)
1432
1433         # step 1 items (networks,vdus,...) to be deleted/updated
1434         for item_index, existing_item in enumerate(existing_list):
1435             target_item = next(
1436                 (t for t in target_list if t["id"] == existing_item["id"]),
1437                 None,
1438             )
1439
1440             for target_vim, existing_viminfo in existing_item.get(
1441                 "vim_info", {}
1442             ).items():
1443                 if existing_viminfo is None:
1444                     continue
1445
1446                 if target_item:
1447                     target_viminfo = target_item.get("vim_info", {}).get(target_vim)
1448                 else:
1449                     target_viminfo = None
1450
1451                 if target_viminfo is None:
1452                     # must be deleted
1453                     self._assign_vim(target_vim)
1454                     target_record_id = "{}.{}".format(db_record, existing_item["id"])
1455                     item_ = item
1456
1457                     if target_vim.startswith("sdn"):
1458                         # item must be sdn-net instead of net if target_vim is a sdn
1459                         item_ = "sdn_net"
1460                         target_record_id += ".sdn"
1461
1462                     deployment_info = {
1463                         "action_id": action_id,
1464                         "nsr_id": nsr_id,
1465                         "task_index": task_index,
1466                     }
1467
1468                     diff_items.append(
1469                         {
1470                             "deployment_info": deployment_info,
1471                             "target_id": target_vim,
1472                             "item": item_,
1473                             "action": "DELETE",
1474                             "target_record": f"{db_record}.{item_index}.vim_info.{target_vim}",
1475                             "target_record_id": target_record_id,
1476                         }
1477                     )
1478                     task_index += 1
1479
1480         # step 2 items (networks,vdus,...) to be created
1481         for target_item in target_list:
1482             item_index = -1
1483
1484             for item_index, existing_item in enumerate(existing_list):
1485                 if existing_item["id"] == target_item["id"]:
1486                     break
1487             else:
1488                 item_index += 1
1489                 db_nsr_update[db_path + ".{}".format(item_index)] = target_item
1490                 existing_list.append(target_item)
1491                 existing_item = None
1492
1493             for target_vim, target_viminfo in target_item.get("vim_info", {}).items():
1494                 existing_viminfo = None
1495
1496                 if existing_item:
1497                     existing_viminfo = existing_item.get("vim_info", {}).get(target_vim)
1498
1499                 if existing_viminfo is not None:
1500                     continue
1501
1502                 target_record_id = "{}.{}".format(db_record, target_item["id"])
1503                 item_ = item
1504
1505                 if target_vim.startswith("sdn"):
1506                     # item must be sdn-net instead of net if target_vim is a sdn
1507                     item_ = "sdn_net"
1508                     target_record_id += ".sdn"
1509
1510                 kwargs = {}
1511                 self.logger.warning(
1512                     "ns.calculate_diff_items target_item={}".format(target_item)
1513                 )
1514                 if process_params == Ns._process_vdu_params:
1515                     self.logger.warning(
1516                         "calculate_diff_items self.fs={}".format(self.fs)
1517                     )
1518                     kwargs.update(
1519                         {
1520                             "vnfr_id": vnfr_id,
1521                             "nsr_id": nsr_id,
1522                             "vnfr": vnfr,
1523                             "vdu2cloud_init": vdu2cloud_init,
1524                             "tasks_by_target_record_id": tasks_by_target_record_id,
1525                             "logger": self.logger,
1526                             "db": self.db,
1527                             "fs": self.fs,
1528                             "ro_nsr_public_key": ro_nsr_public_key,
1529                         }
1530                     )
1531                     self.logger.warning("calculate_diff_items kwargs={}".format(kwargs))
1532
1533                 extra_dict = process_params(
1534                     target_item,
1535                     indata,
1536                     target_viminfo,
1537                     target_record_id,
1538                     **kwargs,
1539                 )
1540                 self._assign_vim(target_vim)
1541
1542                 deployment_info = {
1543                     "action_id": action_id,
1544                     "nsr_id": nsr_id,
1545                     "task_index": task_index,
1546                 }
1547
1548                 new_item = {
1549                     "deployment_info": deployment_info,
1550                     "target_id": target_vim,
1551                     "item": item_,
1552                     "action": "CREATE",
1553                     "target_record": f"{db_record}.{item_index}.vim_info.{target_vim}",
1554                     "target_record_id": target_record_id,
1555                     "extra_dict": extra_dict,
1556                     "common_id": target_item.get("common_id", None),
1557                 }
1558                 diff_items.append(new_item)
1559                 tasks_by_target_record_id[target_record_id] = new_item
1560                 task_index += 1
1561
1562                 db_nsr_update[db_path + ".{}".format(item_index)] = target_item
1563
1564         return diff_items, task_index
1565
1566     def calculate_all_differences_to_deploy(
1567         self,
1568         indata,
1569         nsr_id,
1570         db_nsr,
1571         db_vnfrs,
1572         db_ro_nsr,
1573         db_nsr_update,
1574         db_vnfrs_update,
1575         action_id,
1576         tasks_by_target_record_id,
1577     ):
1578         """This method calculates the ordered list of items (`changes_list`)
1579         to be created and deleted.
1580
1581         Args:
1582             indata (Dict[str, Any]): deployment info
1583             nsr_id (str): NSR id
1584             db_nsr: NSR record from DB
1585             db_vnfrs: VNFRS record from DB
1586             db_ro_nsr (Dict[str, Any]): record from "ro_nsrs"
1587             db_nsr_update (Dict[str, Any]): NSR info to update in DB
1588             db_vnfrs_update (Dict[str, Any]): VNFRS info to update in DB
1589             action_id (str): action id
1590             tasks_by_target_record_id (Dict[str, Any]):
1591                 [<target_record_id>, <task>]
1592
1593         Returns:
1594             List: ordered list of items to be created and deleted.
1595         """
1596
1597         task_index = 0
1598         # set list with diffs:
1599         changes_list = []
1600
1601         # NS vld, image and flavor
1602         for item in ["net", "image", "flavor", "affinity-or-anti-affinity-group"]:
1603             self.logger.debug("process NS={} {}".format(nsr_id, item))
1604             diff_items, task_index = self.calculate_diff_items(
1605                 indata=indata,
1606                 db_nsr=db_nsr,
1607                 db_ro_nsr=db_ro_nsr,
1608                 db_nsr_update=db_nsr_update,
1609                 item=item,
1610                 tasks_by_target_record_id=tasks_by_target_record_id,
1611                 action_id=action_id,
1612                 nsr_id=nsr_id,
1613                 task_index=task_index,
1614                 vnfr_id=None,
1615             )
1616             changes_list += diff_items
1617
1618         # VNF vlds and vdus
1619         for vnfr_id, vnfr in db_vnfrs.items():
1620             # vnfr_id need to be set as global variable for among others nested method _process_vdu_params
1621             for item in ["net", "vdu"]:
1622                 self.logger.debug("process VNF={} {}".format(vnfr_id, item))
1623                 diff_items, task_index = self.calculate_diff_items(
1624                     indata=indata,
1625                     db_nsr=db_nsr,
1626                     db_ro_nsr=db_ro_nsr,
1627                     db_nsr_update=db_vnfrs_update[vnfr["_id"]],
1628                     item=item,
1629                     tasks_by_target_record_id=tasks_by_target_record_id,
1630                     action_id=action_id,
1631                     nsr_id=nsr_id,
1632                     task_index=task_index,
1633                     vnfr_id=vnfr_id,
1634                     vnfr=vnfr,
1635                 )
1636                 changes_list += diff_items
1637
1638         return changes_list
1639
1640     def define_all_tasks(
1641         self,
1642         changes_list,
1643         db_new_tasks,
1644         tasks_by_target_record_id,
1645     ):
1646         """Function to create all the task structures obtanied from
1647         the method calculate_all_differences_to_deploy
1648
1649         Args:
1650             changes_list (List): ordered list of items to be created or deleted
1651             db_new_tasks (List): tasks list to be created
1652             action_id (str): action id
1653             tasks_by_target_record_id (Dict[str, Any]):
1654                 [<target_record_id>, <task>]
1655
1656         """
1657
1658         for change in changes_list:
1659             task = Ns._create_task(
1660                 deployment_info=change["deployment_info"],
1661                 target_id=change["target_id"],
1662                 item=change["item"],
1663                 action=change["action"],
1664                 target_record=change["target_record"],
1665                 target_record_id=change["target_record_id"],
1666                 extra_dict=change.get("extra_dict", None),
1667             )
1668
1669             self.logger.warning("ns.define_all_tasks task={}".format(task))
1670             tasks_by_target_record_id[change["target_record_id"]] = task
1671             db_new_tasks.append(task)
1672
1673             if change.get("common_id"):
1674                 task["common_id"] = change["common_id"]
1675
1676     def upload_all_tasks(
1677         self,
1678         db_new_tasks,
1679         now,
1680     ):
1681         """Function to save all tasks in the common DB
1682
1683         Args:
1684             db_new_tasks (List): tasks list to be created
1685             now (time): current time
1686
1687         """
1688
1689         nb_ro_tasks = 0  # for logging
1690
1691         for db_task in db_new_tasks:
1692             target_id = db_task.pop("target_id")
1693             common_id = db_task.get("common_id")
1694
1695             if common_id:
1696                 if self.db.set_one(
1697                     "ro_tasks",
1698                     q_filter={
1699                         "target_id": target_id,
1700                         "tasks.common_id": common_id,
1701                     },
1702                     update_dict={"to_check_at": now, "modified_at": now},
1703                     push={"tasks": db_task},
1704                     fail_on_empty=False,
1705                 ):
1706                     continue
1707
1708             if not self.db.set_one(
1709                 "ro_tasks",
1710                 q_filter={
1711                     "target_id": target_id,
1712                     "tasks.target_record": db_task["target_record"],
1713                 },
1714                 update_dict={"to_check_at": now, "modified_at": now},
1715                 push={"tasks": db_task},
1716                 fail_on_empty=False,
1717             ):
1718                 # Create a ro_task
1719                 self.logger.debug("Updating database, Creating ro_tasks")
1720                 db_ro_task = Ns._create_ro_task(target_id, db_task)
1721                 nb_ro_tasks += 1
1722                 self.db.create("ro_tasks", db_ro_task)
1723
1724         self.logger.debug(
1725             "Created {} ro_tasks; {} tasks - db_new_tasks={}".format(
1726                 nb_ro_tasks, len(db_new_tasks), db_new_tasks
1727             )
1728         )
1729
1730     def upload_recreate_tasks(
1731         self,
1732         db_new_tasks,
1733         now,
1734     ):
1735         """Function to save recreate tasks in the common DB
1736
1737         Args:
1738             db_new_tasks (List): tasks list to be created
1739             now (time): current time
1740
1741         """
1742
1743         nb_ro_tasks = 0  # for logging
1744
1745         for db_task in db_new_tasks:
1746             target_id = db_task.pop("target_id")
1747             self.logger.warning("target_id={} db_task={}".format(target_id, db_task))
1748
1749             action = db_task.get("action", None)
1750
1751             # Create a ro_task
1752             self.logger.debug("Updating database, Creating ro_tasks")
1753             db_ro_task = Ns._create_ro_task(target_id, db_task)
1754
1755             # If DELETE task: the associated created items should be removed
1756             # (except persistent volumes):
1757             if action == "DELETE":
1758                 db_ro_task["vim_info"]["created"] = True
1759                 db_ro_task["vim_info"]["created_items"] = db_task.get(
1760                     "created_items", {}
1761                 )
1762                 db_ro_task["vim_info"]["vim_id"] = db_task.get("vim_id", None)
1763
1764             nb_ro_tasks += 1
1765             self.logger.warning("upload_all_tasks db_ro_task={}".format(db_ro_task))
1766             self.db.create("ro_tasks", db_ro_task)
1767
1768         self.logger.debug(
1769             "Created {} ro_tasks; {} tasks - db_new_tasks={}".format(
1770                 nb_ro_tasks, len(db_new_tasks), db_new_tasks
1771             )
1772         )
1773
1774     def _prepare_created_items_for_healing(
1775         self,
1776         target_id,
1777         existing_vdu,
1778     ):
1779         # Only ports are considered because created volumes are persistent
1780         ports_list = {}
1781         vim_interfaces = existing_vdu["vim_info"][target_id].get("interfaces", [])
1782         for iface in vim_interfaces:
1783             ports_list["port:" + iface["vim_interface_id"]] = True
1784
1785         return ports_list
1786
1787     def _prepare_persistent_volumes_for_healing(
1788         self,
1789         target_id,
1790         existing_vdu,
1791     ):
1792         # The associated volumes of the VM shouldn't be removed
1793         volumes_list = []
1794         vim_details = {}
1795         vim_details_text = existing_vdu["vim_info"][target_id].get("vim_details", None)
1796         if vim_details_text:
1797             vim_details = yaml.safe_load(f"{vim_details_text}")
1798
1799             for vol_id in vim_details.get("os-extended-volumes:volumes_attached", []):
1800                 volumes_list.append(vol_id["id"])
1801
1802         return volumes_list
1803
1804     def prepare_changes_to_recreate(
1805         self,
1806         indata,
1807         nsr_id,
1808         db_nsr,
1809         db_vnfrs,
1810         db_ro_nsr,
1811         action_id,
1812         tasks_by_target_record_id,
1813     ):
1814         """This method will obtain an ordered list of items (`changes_list`)
1815         to be created and deleted to meet the recreate request.
1816         """
1817
1818         self.logger.debug(
1819             "ns.prepare_changes_to_recreate nsr_id={} indata={}".format(nsr_id, indata)
1820         )
1821
1822         task_index = 0
1823         # set list with diffs:
1824         changes_list = []
1825         db_path = self.db_path_map["vdu"]
1826         target_list = indata.get("healVnfData", {})
1827         vdu2cloud_init = indata.get("cloud_init_content") or {}
1828         ro_nsr_public_key = db_ro_nsr["public_key"]
1829
1830         # Check each VNF of the target
1831         for target_vnf in target_list:
1832             # Find this VNF in the list from DB
1833             vnfr_id = target_vnf.get("vnfInstanceId", None)
1834             if vnfr_id:
1835                 existing_vnf = db_vnfrs.get(vnfr_id)
1836                 db_record = "vnfrs:{}:{}".format(vnfr_id, db_path)
1837                 # vim_account_id = existing_vnf.get("vim-account-id", "")
1838
1839             # Check each VDU of this VNF
1840             for target_vdu in target_vnf["additionalParams"].get("vdu", None):
1841                 vdu_name = target_vdu.get("vdu-id", None)
1842                 # For multi instance VDU count-index is mandatory
1843                 # For single session VDU count-indes is 0
1844                 count_index = target_vdu.get("count-index", 0)
1845                 item_index = 0
1846                 existing_instance = None
1847                 for instance in existing_vnf.get("vdur", None):
1848                     if (
1849                         instance["vdu-name"] == vdu_name
1850                         and instance["count-index"] == count_index
1851                     ):
1852                         existing_instance = instance
1853                         break
1854                     else:
1855                         item_index += 1
1856
1857                 target_record_id = "{}.{}".format(db_record, existing_instance["id"])
1858
1859                 # The target VIM is the one already existing in DB to recreate
1860                 for target_vim, target_viminfo in existing_instance.get(
1861                     "vim_info", {}
1862                 ).items():
1863                     # step 1 vdu to be deleted
1864                     self._assign_vim(target_vim)
1865                     deployment_info = {
1866                         "action_id": action_id,
1867                         "nsr_id": nsr_id,
1868                         "task_index": task_index,
1869                     }
1870
1871                     target_record = f"{db_record}.{item_index}.vim_info.{target_vim}"
1872                     created_items = self._prepare_created_items_for_healing(
1873                         target_vim, existing_instance
1874                     )
1875
1876                     volumes_to_hold = self._prepare_persistent_volumes_for_healing(
1877                         target_vim, existing_instance
1878                     )
1879
1880                     # Specific extra params for recreate tasks:
1881                     extra_dict = {
1882                         "created_items": created_items,
1883                         "vim_id": existing_instance["vim-id"],
1884                         "volumes_to_hold": volumes_to_hold,
1885                     }
1886
1887                     changes_list.append(
1888                         {
1889                             "deployment_info": deployment_info,
1890                             "target_id": target_vim,
1891                             "item": "vdu",
1892                             "action": "DELETE",
1893                             "target_record": target_record,
1894                             "target_record_id": target_record_id,
1895                             "extra_dict": extra_dict,
1896                         }
1897                     )
1898                     delete_task_id = f"{action_id}:{task_index}"
1899                     task_index += 1
1900
1901                     # step 2 vdu to be created
1902                     kwargs = {}
1903                     kwargs.update(
1904                         {
1905                             "vnfr_id": vnfr_id,
1906                             "nsr_id": nsr_id,
1907                             "vnfr": existing_vnf,
1908                             "vdu2cloud_init": vdu2cloud_init,
1909                             "tasks_by_target_record_id": tasks_by_target_record_id,
1910                             "logger": self.logger,
1911                             "db": self.db,
1912                             "fs": self.fs,
1913                             "ro_nsr_public_key": ro_nsr_public_key,
1914                         }
1915                     )
1916
1917                     extra_dict = self._process_recreate_vdu_params(
1918                         existing_instance,
1919                         db_nsr,
1920                         target_viminfo,
1921                         target_record_id,
1922                         target_vim,
1923                         **kwargs,
1924                     )
1925
1926                     # The CREATE task depens on the DELETE task
1927                     extra_dict["depends_on"] = [delete_task_id]
1928
1929                     deployment_info = {
1930                         "action_id": action_id,
1931                         "nsr_id": nsr_id,
1932                         "task_index": task_index,
1933                     }
1934                     self._assign_vim(target_vim)
1935
1936                     new_item = {
1937                         "deployment_info": deployment_info,
1938                         "target_id": target_vim,
1939                         "item": "vdu",
1940                         "action": "CREATE",
1941                         "target_record": target_record,
1942                         "target_record_id": target_record_id,
1943                         "extra_dict": extra_dict,
1944                     }
1945                     changes_list.append(new_item)
1946                     tasks_by_target_record_id[target_record_id] = new_item
1947                     task_index += 1
1948
1949         return changes_list
1950
1951     def recreate(self, session, indata, version, nsr_id, *args, **kwargs):
1952         self.logger.debug("ns.recreate nsr_id={} indata={}".format(nsr_id, indata))
1953         # TODO: validate_input(indata, recreate_schema)
1954         action_id = indata.get("action_id", str(uuid4()))
1955         # get current deployment
1956         db_vnfrs = {}  # vnf's info indexed by _id
1957         step = ""
1958         logging_text = "Recreate nsr_id={} action_id={} indata={}".format(
1959             nsr_id, action_id, indata
1960         )
1961         self.logger.debug(logging_text + "Enter")
1962
1963         try:
1964             step = "Getting ns and vnfr record from db"
1965             db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
1966             db_new_tasks = []
1967             tasks_by_target_record_id = {}
1968             # read from db: vnf's of this ns
1969             step = "Getting vnfrs from db"
1970             db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
1971             self.logger.debug("ns.recreate: db_vnfrs_list={}".format(db_vnfrs_list))
1972
1973             if not db_vnfrs_list:
1974                 raise NsException("Cannot obtain associated VNF for ns")
1975
1976             for vnfr in db_vnfrs_list:
1977                 db_vnfrs[vnfr["_id"]] = vnfr
1978
1979             now = time()
1980             db_ro_nsr = self.db.get_one("ro_nsrs", {"_id": nsr_id}, fail_on_empty=False)
1981             self.logger.debug("ns.recreate: db_ro_nsr={}".format(db_ro_nsr))
1982
1983             if not db_ro_nsr:
1984                 db_ro_nsr = self._create_db_ro_nsrs(nsr_id, now)
1985
1986             with self.write_lock:
1987                 # NS
1988                 step = "process NS elements"
1989                 changes_list = self.prepare_changes_to_recreate(
1990                     indata=indata,
1991                     nsr_id=nsr_id,
1992                     db_nsr=db_nsr,
1993                     db_vnfrs=db_vnfrs,
1994                     db_ro_nsr=db_ro_nsr,
1995                     action_id=action_id,
1996                     tasks_by_target_record_id=tasks_by_target_record_id,
1997                 )
1998
1999                 self.define_all_tasks(
2000                     changes_list=changes_list,
2001                     db_new_tasks=db_new_tasks,
2002                     tasks_by_target_record_id=tasks_by_target_record_id,
2003                 )
2004
2005                 # Delete all ro_tasks registered for the targets vdurs (target_record)
2006                 # If task of type CREATE exist then vim will try to get info form deleted VMs.
2007                 # So remove all task related to target record.
2008                 ro_tasks = self.db.get_list("ro_tasks", {"tasks.nsr_id": nsr_id})
2009                 for change in changes_list:
2010                     for ro_task in ro_tasks:
2011                         for task in ro_task["tasks"]:
2012                             if task["target_record"] == change["target_record"]:
2013                                 self.db.del_one(
2014                                     "ro_tasks",
2015                                     q_filter={
2016                                         "_id": ro_task["_id"],
2017                                         "modified_at": ro_task["modified_at"],
2018                                     },
2019                                     fail_on_empty=False,
2020                                 )
2021
2022                 step = "Updating database, Appending tasks to ro_tasks"
2023                 self.upload_recreate_tasks(
2024                     db_new_tasks=db_new_tasks,
2025                     now=now,
2026                 )
2027
2028             self.logger.debug(
2029                 logging_text + "Exit. Created {} tasks".format(len(db_new_tasks))
2030             )
2031
2032             return (
2033                 {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
2034                 action_id,
2035                 True,
2036             )
2037         except Exception as e:
2038             if isinstance(e, (DbException, NsException)):
2039                 self.logger.error(
2040                     logging_text + "Exit Exception while '{}': {}".format(step, e)
2041                 )
2042             else:
2043                 e = traceback_format_exc()
2044                 self.logger.critical(
2045                     logging_text + "Exit Exception while '{}': {}".format(step, e),
2046                     exc_info=True,
2047                 )
2048
2049             raise NsException(e)
2050
2051     def deploy(self, session, indata, version, nsr_id, *args, **kwargs):
2052         self.logger.debug("ns.deploy nsr_id={} indata={}".format(nsr_id, indata))
2053         validate_input(indata, deploy_schema)
2054         action_id = indata.get("action_id", str(uuid4()))
2055         task_index = 0
2056         # get current deployment
2057         db_nsr_update = {}  # update operation on nsrs
2058         db_vnfrs_update = {}
2059         db_vnfrs = {}  # vnf's info indexed by _id
2060         step = ""
2061         logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id)
2062         self.logger.debug(logging_text + "Enter")
2063
2064         try:
2065             step = "Getting ns and vnfr record from db"
2066             db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
2067             self.logger.debug("ns.deploy: db_nsr={}".format(db_nsr))
2068             db_new_tasks = []
2069             tasks_by_target_record_id = {}
2070             # read from db: vnf's of this ns
2071             step = "Getting vnfrs from db"
2072             db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
2073
2074             if not db_vnfrs_list:
2075                 raise NsException("Cannot obtain associated VNF for ns")
2076
2077             for vnfr in db_vnfrs_list:
2078                 db_vnfrs[vnfr["_id"]] = vnfr
2079                 db_vnfrs_update[vnfr["_id"]] = {}
2080             self.logger.debug("ns.deploy db_vnfrs={}".format(db_vnfrs))
2081
2082             now = time()
2083             db_ro_nsr = self.db.get_one("ro_nsrs", {"_id": nsr_id}, fail_on_empty=False)
2084
2085             if not db_ro_nsr:
2086                 db_ro_nsr = self._create_db_ro_nsrs(nsr_id, now)
2087
2088             # check that action_id is not in the list of actions. Suffixed with :index
2089             if action_id in db_ro_nsr["actions"]:
2090                 index = 1
2091
2092                 while True:
2093                     new_action_id = "{}:{}".format(action_id, index)
2094
2095                     if new_action_id not in db_ro_nsr["actions"]:
2096                         action_id = new_action_id
2097                         self.logger.debug(
2098                             logging_text
2099                             + "Changing action_id in use to {}".format(action_id)
2100                         )
2101                         break
2102
2103                     index += 1
2104
2105             def _process_action(indata):
2106                 nonlocal db_new_tasks
2107                 nonlocal action_id
2108                 nonlocal nsr_id
2109                 nonlocal task_index
2110                 nonlocal db_vnfrs
2111                 nonlocal db_ro_nsr
2112
2113                 if indata["action"]["action"] == "inject_ssh_key":
2114                     key = indata["action"].get("key")
2115                     user = indata["action"].get("user")
2116                     password = indata["action"].get("password")
2117
2118                     for vnf in indata.get("vnf", ()):
2119                         if vnf["_id"] not in db_vnfrs:
2120                             raise NsException("Invalid vnf={}".format(vnf["_id"]))
2121
2122                         db_vnfr = db_vnfrs[vnf["_id"]]
2123
2124                         for target_vdu in vnf.get("vdur", ()):
2125                             vdu_index, vdur = next(
2126                                 (
2127                                     i_v
2128                                     for i_v in enumerate(db_vnfr["vdur"])
2129                                     if i_v[1]["id"] == target_vdu["id"]
2130                                 ),
2131                                 (None, None),
2132                             )
2133
2134                             if not vdur:
2135                                 raise NsException(
2136                                     "Invalid vdu vnf={}.{}".format(
2137                                         vnf["_id"], target_vdu["id"]
2138                                     )
2139                                 )
2140
2141                             target_vim, vim_info = next(
2142                                 k_v for k_v in vdur["vim_info"].items()
2143                             )
2144                             self._assign_vim(target_vim)
2145                             target_record = "vnfrs:{}:vdur.{}.ssh_keys".format(
2146                                 vnf["_id"], vdu_index
2147                             )
2148                             extra_dict = {
2149                                 "depends_on": [
2150                                     "vnfrs:{}:vdur.{}".format(vnf["_id"], vdur["id"])
2151                                 ],
2152                                 "params": {
2153                                     "ip_address": vdur.get("ip-address"),
2154                                     "user": user,
2155                                     "key": key,
2156                                     "password": password,
2157                                     "private_key": db_ro_nsr["private_key"],
2158                                     "salt": db_ro_nsr["_id"],
2159                                     "schema_version": db_ro_nsr["_admin"][
2160                                         "schema_version"
2161                                     ],
2162                                 },
2163                             }
2164
2165                             deployment_info = {
2166                                 "action_id": action_id,
2167                                 "nsr_id": nsr_id,
2168                                 "task_index": task_index,
2169                             }
2170
2171                             task = Ns._create_task(
2172                                 deployment_info=deployment_info,
2173                                 target_id=target_vim,
2174                                 item="vdu",
2175                                 action="EXEC",
2176                                 target_record=target_record,
2177                                 target_record_id=None,
2178                                 extra_dict=extra_dict,
2179                             )
2180
2181                             task_index = deployment_info.get("task_index")
2182
2183                             db_new_tasks.append(task)
2184
2185             with self.write_lock:
2186                 if indata.get("action"):
2187                     _process_action(indata)
2188                 else:
2189                     # compute network differences
2190                     # NS
2191                     step = "process NS elements"
2192                     changes_list = self.calculate_all_differences_to_deploy(
2193                         indata=indata,
2194                         nsr_id=nsr_id,
2195                         db_nsr=db_nsr,
2196                         db_vnfrs=db_vnfrs,
2197                         db_ro_nsr=db_ro_nsr,
2198                         db_nsr_update=db_nsr_update,
2199                         db_vnfrs_update=db_vnfrs_update,
2200                         action_id=action_id,
2201                         tasks_by_target_record_id=tasks_by_target_record_id,
2202                     )
2203                     self.define_all_tasks(
2204                         changes_list=changes_list,
2205                         db_new_tasks=db_new_tasks,
2206                         tasks_by_target_record_id=tasks_by_target_record_id,
2207                     )
2208
2209                 step = "Updating database, Appending tasks to ro_tasks"
2210                 self.upload_all_tasks(
2211                     db_new_tasks=db_new_tasks,
2212                     now=now,
2213                 )
2214
2215                 step = "Updating database, nsrs"
2216                 if db_nsr_update:
2217                     self.db.set_one("nsrs", {"_id": nsr_id}, db_nsr_update)
2218
2219                 for vnfr_id, db_vnfr_update in db_vnfrs_update.items():
2220                     if db_vnfr_update:
2221                         step = "Updating database, vnfrs={}".format(vnfr_id)
2222                         self.db.set_one("vnfrs", {"_id": vnfr_id}, db_vnfr_update)
2223
2224             self.logger.debug(
2225                 logging_text + "Exit. Created {} tasks".format(len(db_new_tasks))
2226             )
2227
2228             return (
2229                 {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
2230                 action_id,
2231                 True,
2232             )
2233         except Exception as e:
2234             if isinstance(e, (DbException, NsException)):
2235                 self.logger.error(
2236                     logging_text + "Exit Exception while '{}': {}".format(step, e)
2237                 )
2238             else:
2239                 e = traceback_format_exc()
2240                 self.logger.critical(
2241                     logging_text + "Exit Exception while '{}': {}".format(step, e),
2242                     exc_info=True,
2243                 )
2244
2245             raise NsException(e)
2246
2247     def delete(self, session, indata, version, nsr_id, *args, **kwargs):
2248         self.logger.debug("ns.delete version={} nsr_id={}".format(version, nsr_id))
2249         # self.db.del_list({"_id": ro_task["_id"], "tasks.nsr_id.ne": nsr_id})
2250
2251         with self.write_lock:
2252             try:
2253                 NsWorker.delete_db_tasks(self.db, nsr_id, None)
2254             except NsWorkerException as e:
2255                 raise NsException(e)
2256
2257         return None, None, True
2258
2259     def status(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
2260         self.logger.debug(
2261             "ns.status version={} nsr_id={}, action_id={} indata={}".format(
2262                 version, nsr_id, action_id, indata
2263             )
2264         )
2265         task_list = []
2266         done = 0
2267         total = 0
2268         ro_tasks = self.db.get_list("ro_tasks", {"tasks.action_id": action_id})
2269         global_status = "DONE"
2270         details = []
2271
2272         for ro_task in ro_tasks:
2273             for task in ro_task["tasks"]:
2274                 if task and task["action_id"] == action_id:
2275                     task_list.append(task)
2276                     total += 1
2277
2278                     if task["status"] == "FAILED":
2279                         global_status = "FAILED"
2280                         error_text = "Error at {} {}: {}".format(
2281                             task["action"].lower(),
2282                             task["item"],
2283                             ro_task["vim_info"].get("vim_details") or "unknown",
2284                         )
2285                         details.append(error_text)
2286                     elif task["status"] in ("SCHEDULED", "BUILD"):
2287                         if global_status != "FAILED":
2288                             global_status = "BUILD"
2289                     else:
2290                         done += 1
2291
2292         return_data = {
2293             "status": global_status,
2294             "details": ". ".join(details)
2295             if details
2296             else "progress {}/{}".format(done, total),
2297             "nsr_id": nsr_id,
2298             "action_id": action_id,
2299             "tasks": task_list,
2300         }
2301
2302         return return_data, None, True
2303
2304     def recreate_status(
2305         self, session, indata, version, nsr_id, action_id, *args, **kwargs
2306     ):
2307         return self.status(session, indata, version, nsr_id, action_id, *args, **kwargs)
2308
2309     def cancel(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
2310         print(
2311             "ns.cancel session={} indata={} version={} nsr_id={}, action_id={}".format(
2312                 session, indata, version, nsr_id, action_id
2313             )
2314         )
2315
2316         return None, None, True
2317
2318     def get_deploy(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
2319         nsrs = self.db.get_list("nsrs", {})
2320         return_data = []
2321
2322         for ns in nsrs:
2323             return_data.append({"_id": ns["_id"], "name": ns["name"]})
2324
2325         return return_data, None, True
2326
2327     def get_actions(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
2328         ro_tasks = self.db.get_list("ro_tasks", {"tasks.nsr_id": nsr_id})
2329         return_data = []
2330
2331         for ro_task in ro_tasks:
2332             for task in ro_task["tasks"]:
2333                 if task["action_id"] not in return_data:
2334                     return_data.append(task["action_id"])
2335
2336         return return_data, None, True