osm_lcm/lcm_utils.py

   1 # -*- coding: utf-8 -*-
   2
   3 ##
   4 # Copyright 2018 Telefonica S.A.
   5 #
   6 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   7 # not use this file except in compliance with the License. You may obtain
   8 # a copy of the License at
   9 #
  10 #         http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 # Unless required by applicable law or agreed to in writing, software
  13 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  14 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  15 # License for the specific language governing permissions and limitations
  16 # under the License.
  17 ##
  18
  19 import asyncio
  20 import checksumdir
  21 from collections import OrderedDict
  22 import hashlib
  23 import os
  24 import shutil
  25 import traceback
  26 from time import time
  27
  28 from osm_common.fsbase import FsException
  29 from osm_lcm.data_utils.database.database import Database
  30 from osm_lcm.data_utils.filesystem.filesystem import Filesystem
  31 import yaml
  32 from zipfile import ZipFile, BadZipfile
  33
  34 # from osm_common.dbbase import DbException
  35
  36 __author__ = "Alfonso Tierno"
  37
  38
  39 class LcmException(Exception):
  40     pass
  41
  42
  43 class LcmExceptionExit(LcmException):
  44     pass
  45
  46
  47 def versiontuple(v):
  48     """utility for compare dot separate versions. Fills with zeros to proper number comparison
  49     package version will be something like 4.0.1.post11+gb3f024d.dirty-1. Where 4.0.1 is the git tag, postXX is the
  50     number of commits from this tag, and +XXXXXXX is the git commit short id. Total length is 16 with until 999 commits
  51     """
  52     filled = []
  53     for point in v.split("."):
  54         point, _, _ = point.partition("+")
  55         point, _, _ = point.partition("-")
  56         filled.append(point.zfill(20))
  57     return tuple(filled)
  58
  59
  60 def deep_get(target_dict, key_list, default_value=None):
  61     """
  62     Get a value from target_dict entering in the nested keys. If keys does not exist, it returns None
  63     Example target_dict={a: {b: 5}}; key_list=[a,b] returns 5; both key_list=[a,b,c] and key_list=[f,h] return None
  64     :param target_dict: dictionary to be read
  65     :param key_list: list of keys to read from  target_dict
  66     :param default_value: value to return if key is not present in the nested dictionary
  67     :return: The wanted value if exist, None otherwise
  68     """
  69     for key in key_list:
  70         if not isinstance(target_dict, dict) or key not in target_dict:
  71             return default_value
  72         target_dict = target_dict[key]
  73     return target_dict
  74
  75
  76 def get_iterable(in_dict, in_key):
  77     """
  78     Similar to <dict>.get(), but if value is None, False, ..., An empty tuple is returned instead
  79     :param in_dict: a dictionary
  80     :param in_key: the key to look for at in_dict
  81     :return: in_dict[in_var] or () if it is None or not present
  82     """
  83     if not in_dict.get(in_key):
  84         return ()
  85     return in_dict[in_key]
  86
  87
  88 def check_juju_bundle_existence(vnfd: dict) -> str:
  89     """Checks the existence of juju-bundle in the descriptor
  90
  91     Args:
  92         vnfd:   Descriptor as a dictionary
  93
  94     Returns:
  95         Juju bundle if dictionary has juju-bundle else None
  96
  97     """
  98     if vnfd.get("vnfd"):
  99         vnfd = vnfd["vnfd"]
 100
 101     for kdu in vnfd.get("kdu", []):
 102         return kdu.get("juju-bundle", None)
 103
 104
 105 def get_charm_artifact_path(base_folder, charm_name, charm_type, revision=str()) -> str:
 106     """Finds the charm artifact paths
 107
 108     Args:
 109         base_folder:    Main folder which will be looked up for charm
 110         charm_name:   Charm name
 111         charm_type:   Type of charm native_charm, lxc_proxy_charm or k8s_proxy_charm
 112         revision:   vnf package revision number if there is
 113
 114     Returns:
 115         artifact_path: (str)
 116
 117     """
 118     extension = ""
 119     if revision:
 120         extension = ":" + str(revision)
 121
 122     if base_folder.get("pkg-dir"):
 123         artifact_path = "{}/{}/{}/{}".format(
 124             base_folder["folder"].split(":")[0] + extension,
 125             base_folder["pkg-dir"],
 126             "charms"
 127             if charm_type in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
 128             else "helm-charts",
 129             charm_name,
 130         )
 131
 132     else:
 133         # For SOL004 packages
 134         artifact_path = "{}/Scripts/{}/{}".format(
 135             base_folder["folder"].split(":")[0] + extension,
 136             "charms"
 137             if charm_type in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
 138             else "helm-charts",
 139             charm_name,
 140         )
 141
 142     return artifact_path
 143
 144
 145 def populate_dict(target_dict, key_list, value):
 146     """
 147     Update target_dict creating nested dictionaries with the key_list. Last key_list item is asigned the value.
 148     Example target_dict={K: J}; key_list=[a,b,c];  target_dict will be {K: J, a: {b: {c: value}}}
 149     :param target_dict: dictionary to be changed
 150     :param key_list: list of keys to insert at target_dict
 151     :param value:
 152     :return: None
 153     """
 154     for key in key_list[0:-1]:
 155         if key not in target_dict:
 156             target_dict[key] = {}
 157         target_dict = target_dict[key]
 158     target_dict[key_list[-1]] = value
 159
 160
 161 def get_ee_id_parts(ee_id):
 162     """
 163     Parses ee_id stored at database that can be either 'version:namespace.helm_id' or only
 164     namespace.helm_id for backward compatibility
 165     If exists helm version can be helm-v3 or helm (helm-v2 old version)
 166     """
 167     version, _, part_id = ee_id.rpartition(":")
 168     namespace, _, helm_id = part_id.rpartition(".")
 169     return version, namespace, helm_id
 170
 171
 172 def vld_to_ro_ip_profile(source_data):
 173     if source_data:
 174         return {
 175             "ip_version": "IPv4"
 176             if "v4" in source_data.get("ip-version", "ipv4")
 177             else "IPv6",
 178             "subnet_address": source_data.get("cidr")
 179             or source_data.get("subnet-address"),
 180             "gateway_address": source_data.get("gateway-ip")
 181             or source_data.get("gateway-address"),
 182             "dns_address": ";".join(
 183                 [v["address"] for v in source_data["dns-server"] if v.get("address")]
 184             )
 185             if source_data.get("dns-server")
 186             else None,
 187             "dhcp_enabled": source_data.get("dhcp-params", {}).get("enabled", False)
 188             or source_data.get("dhcp-enabled", False),
 189             "dhcp_start_address": source_data["dhcp-params"].get("start-address")
 190             if source_data.get("dhcp-params")
 191             else None,
 192             "dhcp_count": source_data["dhcp-params"].get("count")
 193             if source_data.get("dhcp-params")
 194             else None,
 195             "ipv6_address_mode": source_data["ipv6-address-mode"]
 196             if "ipv6-address-mode" in source_data
 197             else None,
 198         }
 199
 200
 201 class LcmBase:
 202     def __init__(self, msg, logger):
 203         """
 204
 205         :param db: database connection
 206         """
 207         self.db = Database().instance.db
 208         self.msg = msg
 209         self.fs = Filesystem().instance.fs
 210         self.logger = logger
 211
 212     def update_db_2(self, item, _id, _desc):
 213         """
 214         Updates database with _desc information. If success _desc is cleared
 215         :param item: collection
 216         :param _id: the _id to use in the query filter
 217         :param _desc: dictionary with the content to update. Keys are dot separated keys for
 218         :return: None. Exception is raised on error
 219         """
 220         if not _desc:
 221             return
 222         now = time()
 223         _desc["_admin.modified"] = now
 224         self.db.set_one(item, {"_id": _id}, _desc)
 225         _desc.clear()
 226         # except DbException as e:
 227         #     self.logger.error("Updating {} _id={} with '{}'. Error: {}".format(item, _id, _desc, e))
 228
 229     @staticmethod
 230     def calculate_charm_hash(zipped_file):
 231         """Calculate the hash of charm files which ends with .charm
 232
 233         Args:
 234             zipped_file (str): Existing charm package full path
 235
 236         Returns:
 237             hex digest  (str): The hash of the charm file
 238         """
 239         filehash = hashlib.sha256()
 240         with open(zipped_file, mode="rb") as file:
 241             contents = file.read()
 242             filehash.update(contents)
 243             return filehash.hexdigest()
 244
 245     @staticmethod
 246     def compare_charm_hash(current_charm, target_charm):
 247         """Compare the existing charm and the target charm if the charms
 248         are given as zip files ends with .charm
 249
 250         Args:
 251             current_charm (str): Existing charm package full path
 252             target_charm  (str): Target charm package full path
 253
 254         Returns:
 255             True/False (bool): if charm has changed it returns True
 256         """
 257         return LcmBase.calculate_charm_hash(
 258             current_charm
 259         ) != LcmBase.calculate_charm_hash(target_charm)
 260
 261     @staticmethod
 262     def compare_charmdir_hash(current_charm_dir, target_charm_dir):
 263         """Compare the existing charm and the target charm if the charms
 264         are given as directories
 265
 266         Args:
 267             current_charm_dir (str): Existing charm package directory path
 268             target_charm_dir  (str): Target charm package directory path
 269
 270         Returns:
 271             True/False (bool): if charm has changed it returns True
 272         """
 273         return checksumdir.dirhash(current_charm_dir) != checksumdir.dirhash(
 274             target_charm_dir
 275         )
 276
 277     def check_charm_hash_changed(
 278         self, current_charm_path: str, target_charm_path: str
 279     ) -> bool:
 280         """Find the target charm has changed or not by checking the hash of
 281         old and new charm packages
 282
 283         Args:
 284             current_charm_path (str): Existing charm package artifact path
 285             target_charm_path  (str): Target charm package artifact path
 286
 287         Returns:
 288             True/False (bool): if charm has changed it returns True
 289
 290         """
 291         try:
 292             # Check if the charm artifacts are available
 293             current_charm = self.fs.path + current_charm_path
 294             target_charm = self.fs.path + target_charm_path
 295
 296             if os.path.exists(current_charm) and os.path.exists(target_charm):
 297                 # Compare the hash of .charm files
 298                 if current_charm.endswith(".charm"):
 299                     return LcmBase.compare_charm_hash(current_charm, target_charm)
 300
 301                 # Compare the hash of charm folders
 302                 return LcmBase.compare_charmdir_hash(current_charm, target_charm)
 303
 304             else:
 305                 raise LcmException(
 306                     "Charm artifact {} does not exist in the VNF Package".format(
 307                         self.fs.path + target_charm_path
 308                     )
 309                 )
 310         except (IOError, OSError, TypeError) as error:
 311             self.logger.debug(traceback.format_exc())
 312             self.logger.error(f"{error} occured while checking the charm hashes")
 313             raise LcmException(error)
 314
 315     @staticmethod
 316     def get_charm_name(charm_metadata_file: str) -> str:
 317         """Get the charm name from metadata file.
 318
 319         Args:
 320             charm_metadata_file    (str):  charm metadata file full path
 321
 322         Returns:
 323             charm_name    (str):  charm name
 324
 325         """
 326         # Read charm metadata.yaml to get the charm name
 327         with open(charm_metadata_file, "r") as metadata_file:
 328             content = yaml.safe_load(metadata_file)
 329             charm_name = content["name"]
 330             return str(charm_name)
 331
 332     def _get_charm_path(
 333         self, nsd_package_path: str, nsd_package_name: str, charm_folder_name: str
 334     ) -> str:
 335         """Get the full path of charm folder.
 336
 337         Args:
 338             nsd_package_path    (str):  NSD package full path
 339             nsd_package_name    (str):  NSD package name
 340             charm_folder_name   (str):  folder name
 341
 342         Returns:
 343             charm_path    (str):  charm folder full path
 344         """
 345         charm_path = (
 346             self.fs.path
 347             + nsd_package_path
 348             + "/"
 349             + nsd_package_name
 350             + "/charms/"
 351             + charm_folder_name
 352         )
 353         return charm_path
 354
 355     def _get_charm_metadata_file(
 356         self,
 357         charm_folder_name: str,
 358         nsd_package_path: str,
 359         nsd_package_name: str,
 360         charm_path: str = None,
 361     ) -> str:
 362         """Get the path of charm metadata file.
 363
 364         Args:
 365             charm_folder_name   (str):  folder name
 366             nsd_package_path    (str):  NSD package full path
 367             nsd_package_name    (str):  NSD package name
 368             charm_path  (str):  Charm full path
 369
 370         Returns:
 371             charm_metadata_file_path    (str):  charm metadata file full path
 372
 373         """
 374         # Locate the charm metadata.yaml
 375         if charm_folder_name.endswith(".charm"):
 376             extract_path = (
 377                 self.fs.path
 378                 + nsd_package_path
 379                 + "/"
 380                 + nsd_package_name
 381                 + "/charms/"
 382                 + charm_folder_name.replace(".charm", "")
 383             )
 384             # Extract .charm to extract path
 385             with ZipFile(charm_path, "r") as zipfile:
 386                 zipfile.extractall(extract_path)
 387             return extract_path + "/metadata.yaml"
 388         else:
 389             return charm_path + "/metadata.yaml"
 390
 391     def find_charm_name(self, db_nsr: dict, charm_folder_name: str) -> str:
 392         """Get the charm name from metadata.yaml of charm package.
 393
 394         Args:
 395             db_nsr  (dict): NS record as a dictionary
 396             charm_folder_name   (str): charm folder name
 397
 398         Returns:
 399              charm_name (str):  charm name
 400         """
 401         try:
 402             if not charm_folder_name:
 403                 raise LcmException("charm_folder_name should be provided.")
 404
 405             # Find nsd_package details: path, name
 406             revision = db_nsr.get("revision", "")
 407
 408             # Get the NSD package path
 409             if revision:
 410                 nsd_package_path = db_nsr["nsd-id"] + ":" + str(revision)
 411                 db_nsd = self.db.get_one("nsds_revisions", {"_id": nsd_package_path})
 412
 413             else:
 414                 nsd_package_path = db_nsr["nsd-id"]
 415
 416                 db_nsd = self.db.get_one("nsds", {"_id": nsd_package_path})
 417
 418             # Get the NSD package name
 419             nsd_package_name = db_nsd["_admin"]["storage"]["pkg-dir"]
 420
 421             # Remove the existing nsd package and sync from FsMongo
 422             shutil.rmtree(self.fs.path + nsd_package_path, ignore_errors=True)
 423             self.fs.sync(from_path=nsd_package_path)
 424
 425             # Get the charm path
 426             charm_path = self._get_charm_path(
 427                 nsd_package_path, nsd_package_name, charm_folder_name
 428             )
 429
 430             # Find charm metadata file full path
 431             charm_metadata_file = self._get_charm_metadata_file(
 432                 charm_folder_name, nsd_package_path, nsd_package_name, charm_path
 433             )
 434
 435             # Return charm name
 436             return self.get_charm_name(charm_metadata_file)
 437
 438         except (
 439             yaml.YAMLError,
 440             IOError,
 441             FsException,
 442             KeyError,
 443             TypeError,
 444             FileNotFoundError,
 445             BadZipfile,
 446         ) as error:
 447             self.logger.debug(traceback.format_exc())
 448             self.logger.error(f"{error} occured while getting the charm name")
 449             raise LcmException(error)
 450
 451     def get_vca_info(self, ee_item, db_nsr, get_charm_name: bool):
 452         vca_name = charm_name = vca_type = None
 453         if ee_item.get("juju"):
 454             vca_name = ee_item["juju"].get("charm")
 455             if get_charm_name:
 456                 charm_name = self.find_charm_name(db_nsr, str(vca_name))
 457             vca_type = (
 458                 "lxc_proxy_charm"
 459                 if ee_item["juju"].get("charm") is not None
 460                 else "native_charm"
 461             )
 462             if ee_item["juju"].get("cloud") == "k8s":
 463                 vca_type = "k8s_proxy_charm"
 464             elif ee_item["juju"].get("proxy") is False:
 465                 vca_type = "native_charm"
 466         elif ee_item.get("helm-chart"):
 467             vca_name = ee_item["helm-chart"]
 468             vca_type = "helm-v3"
 469         return vca_name, charm_name, vca_type
 470
 471
 472 class TaskRegistry(LcmBase):
 473     """
 474     Implements a registry of task needed for later cancelation, look for related tasks that must be completed before
 475     etc. It stores a four level dict
 476     First level is the topic, ns, vim_account, sdn
 477     Second level is the _id
 478     Third level is the operation id
 479     Fourth level is a descriptive name, the value is the task class
 480
 481     The HA (High-Availability) methods are used when more than one LCM instance is running.
 482     To register the current task in the external DB, use LcmBase as base class, to be able
 483     to reuse LcmBase.update_db_2()
 484     The DB registry uses the following fields to distinguish a task:
 485     - op_type: operation type ("nslcmops" or "nsilcmops")
 486     - op_id:   operation ID
 487     - worker:  the worker ID for this process
 488     """
 489
 490     # NS/NSI: "services" VIM/WIM/SDN: "accounts"
 491     topic_service_list = ["ns", "nsi"]
 492     topic_account_list = ["vim", "wim", "sdn", "k8scluster", "vca", "k8srepo"]
 493
 494     # Map topic to InstanceID
 495     topic2instid_dict = {"ns": "nsInstanceId", "nsi": "netsliceInstanceId"}
 496
 497     # Map topic to DB table name
 498     topic2dbtable_dict = {
 499         "ns": "nslcmops",
 500         "nsi": "nsilcmops",
 501         "vim": "vim_accounts",
 502         "wim": "wim_accounts",
 503         "sdn": "sdns",
 504         "k8scluster": "k8sclusters",
 505         "vca": "vca",
 506         "k8srepo": "k8srepos",
 507     }
 508
 509     def __init__(self, worker_id=None, logger=None):
 510         self.task_registry = {
 511             "ns": {},
 512             "nsi": {},
 513             "vim_account": {},
 514             "wim_account": {},
 515             "sdn": {},
 516             "k8scluster": {},
 517             "vca": {},
 518             "k8srepo": {},
 519         }
 520         self.worker_id = worker_id
 521         self.db = Database().instance.db
 522         self.logger = logger
 523
 524     def register(self, topic, _id, op_id, task_name, task):
 525         """
 526         Register a new task
 527         :param topic: Can be "ns", "nsi", "vim_account", "sdn"
 528         :param _id: _id of the related item
 529         :param op_id: id of the operation of the related item
 530         :param task_name: Task descriptive name, as create, instantiate, terminate. Must be unique in this op_id
 531         :param task: Task class
 532         :return: none
 533         """
 534         if _id not in self.task_registry[topic]:
 535             self.task_registry[topic][_id] = OrderedDict()
 536         if op_id not in self.task_registry[topic][_id]:
 537             self.task_registry[topic][_id][op_id] = {task_name: task}
 538         else:
 539             self.task_registry[topic][_id][op_id][task_name] = task
 540         # print("registering task", topic, _id, op_id, task_name, task)
 541
 542     def remove(self, topic, _id, op_id, task_name=None):
 543         """
 544         When task is ended, it should be removed. It ignores missing tasks. It also removes tasks done with this _id
 545         :param topic: Can be "ns", "nsi", "vim_account", "sdn"
 546         :param _id: _id of the related item
 547         :param op_id: id of the operation of the related item
 548         :param task_name: Task descriptive name. If none it deletes all tasks with same _id and op_id
 549         :return: None
 550         """
 551         if not self.task_registry[topic].get(_id):
 552             return
 553         if not task_name:
 554             self.task_registry[topic][_id].pop(op_id, None)
 555         elif self.task_registry[topic][_id].get(op_id):
 556             self.task_registry[topic][_id][op_id].pop(task_name, None)
 557
 558         # delete done tasks
 559         for op_id_ in list(self.task_registry[topic][_id]):
 560             for name, task in self.task_registry[topic][_id][op_id_].items():
 561                 if not task.done():
 562                     break
 563             else:
 564                 del self.task_registry[topic][_id][op_id_]
 565         if not self.task_registry[topic][_id]:
 566             del self.task_registry[topic][_id]
 567
 568     def lookfor_related(self, topic, _id, my_op_id=None):
 569         task_list = []
 570         task_name_list = []
 571         if _id not in self.task_registry[topic]:
 572             return "", task_name_list
 573         for op_id in reversed(self.task_registry[topic][_id]):
 574             if my_op_id:
 575                 if my_op_id == op_id:
 576                     my_op_id = None  # so that the next task is taken
 577                 continue
 578
 579             for task_name, task in self.task_registry[topic][_id][op_id].items():
 580                 if not task.done():
 581                     task_list.append(task)
 582                     task_name_list.append(task_name)
 583             break
 584         return ", ".join(task_name_list), task_list
 585
 586     def cancel(self, topic, _id, target_op_id=None, target_task_name=None):
 587         """
 588         Cancel all active tasks of a concrete ns, nsi, vim_account, sdn identified for _id. If op_id is supplied only
 589         this is cancelled, and the same with task_name
 590         :return: cancelled task to be awaited if needed
 591         """
 592         if not self.task_registry[topic].get(_id):
 593             return
 594         for op_id in reversed(self.task_registry[topic][_id]):
 595             if target_op_id and target_op_id != op_id:
 596                 continue
 597             for task_name, task in list(self.task_registry[topic][_id][op_id].items()):
 598                 if target_task_name and target_task_name != task_name:
 599                     continue
 600                 # result =
 601                 task.cancel()
 602                 yield task
 603                 # if result:
 604                 #     self.logger.debug("{} _id={} order_id={} task={} cancelled".format(topic, _id, op_id, task_name))
 605
 606     # Is topic NS/NSI?
 607     def _is_service_type_HA(self, topic):
 608         return topic in self.topic_service_list
 609
 610     # Is topic VIM/WIM/SDN?
 611     def _is_account_type_HA(self, topic):
 612         return topic in self.topic_account_list
 613
 614     # Input: op_id, example: 'abc123def:3' Output: account_id='abc123def', op_index=3
 615     def _get_account_and_op_HA(self, op_id):
 616         if not op_id:
 617             return None, None
 618         account_id, _, op_index = op_id.rpartition(":")
 619         if not account_id or not op_index.isdigit():
 620             return None, None
 621         return account_id, op_index
 622
 623     # Get '_id' for any topic and operation
 624     def _get_instance_id_HA(self, topic, op_type, op_id):
 625         _id = None
 626         # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
 627         if op_type == "ANY":
 628             _id = op_id
 629         # NS/NSI: Use op_id as '_id'
 630         elif self._is_service_type_HA(topic):
 631             _id = op_id
 632         # VIM/SDN/WIM/K8SCLUSTER: Split op_id to get Account ID and Operation Index, use Account ID as '_id'
 633         elif self._is_account_type_HA(topic):
 634             _id, _ = self._get_account_and_op_HA(op_id)
 635         return _id
 636
 637     # Set DB _filter for querying any related process state
 638     def _get_waitfor_filter_HA(self, db_lcmop, topic, op_type, op_id):
 639         _filter = {}
 640         # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
 641         # In this special case, the timestamp is ignored
 642         if op_type == "ANY":
 643             _filter = {"operationState": "PROCESSING"}
 644         # Otherwise, get 'startTime' timestamp for this operation
 645         else:
 646             # NS/NSI
 647             if self._is_service_type_HA(topic):
 648                 now = time()
 649                 starttime_this_op = db_lcmop.get("startTime")
 650                 instance_id_label = self.topic2instid_dict.get(topic)
 651                 instance_id = db_lcmop.get(instance_id_label)
 652                 _filter = {
 653                     instance_id_label: instance_id,
 654                     "operationState": "PROCESSING",
 655                     "startTime.lt": starttime_this_op,
 656                     "_admin.modified.gt": now
 657                     - 2 * 3600,  # ignore if tow hours of inactivity
 658                 }
 659             # VIM/WIM/SDN/K8scluster
 660             elif self._is_account_type_HA(topic):
 661                 _, op_index = self._get_account_and_op_HA(op_id)
 662                 _ops = db_lcmop["_admin"]["operations"]
 663                 _this_op = _ops[int(op_index)]
 664                 starttime_this_op = _this_op.get("startTime", None)
 665                 _filter = {
 666                     "operationState": "PROCESSING",
 667                     "startTime.lt": starttime_this_op,
 668                 }
 669         return _filter
 670
 671     # Get DB params for any topic and operation
 672     def _get_dbparams_for_lock_HA(self, topic, op_type, op_id):
 673         q_filter = {}
 674         update_dict = {}
 675         # NS/NSI
 676         if self._is_service_type_HA(topic):
 677             q_filter = {"_id": op_id, "_admin.worker": None}
 678             update_dict = {"_admin.worker": self.worker_id}
 679         # VIM/WIM/SDN
 680         elif self._is_account_type_HA(topic):
 681             account_id, op_index = self._get_account_and_op_HA(op_id)
 682             if not account_id:
 683                 return None, None
 684             if op_type == "create":
 685                 # Creating a VIM/WIM/SDN account implies setting '_admin.current_operation' = 0
 686                 op_index = 0
 687             q_filter = {
 688                 "_id": account_id,
 689                 "_admin.operations.{}.worker".format(op_index): None,
 690             }
 691             update_dict = {
 692                 "_admin.operations.{}.worker".format(op_index): self.worker_id,
 693                 "_admin.current_operation": op_index,
 694             }
 695         return q_filter, update_dict
 696
 697     def lock_HA(self, topic, op_type, op_id):
 698         """
 699         Lock a task, if possible, to indicate to the HA system that
 700         the task will be executed in this LCM instance.
 701         :param topic: Can be "ns", "nsi", "vim", "wim", or "sdn"
 702         :param op_type: Operation type, can be "nslcmops", "nsilcmops", "create", "edit", "delete"
 703         :param op_id: NS, NSI: Operation ID  VIM,WIM,SDN: Account ID + ':' + Operation Index
 704         :return:
 705         True=lock was successful => execute the task (not registered by any other LCM instance)
 706         False=lock failed => do NOT execute the task (already registered by another LCM instance)
 707
 708         HA tasks and backward compatibility:
 709         If topic is "account type" (VIM/WIM/SDN) and op_id is None, 'op_id' was not provided by NBI.
 710         This means that the running NBI instance does not support HA.
 711         In such a case this method should always return True, to always execute
 712         the task in this instance of LCM, without querying the DB.
 713         """
 714
 715         # Backward compatibility for VIM/WIM/SDN/k8scluster without op_id
 716         if self._is_account_type_HA(topic) and op_id is None:
 717             return True
 718
 719         # Try to lock this task
 720         db_table_name = self.topic2dbtable_dict[topic]
 721         q_filter, update_dict = self._get_dbparams_for_lock_HA(topic, op_type, op_id)
 722         db_lock_task = self.db.set_one(
 723             db_table_name,
 724             q_filter=q_filter,
 725             update_dict=update_dict,
 726             fail_on_empty=False,
 727         )
 728         if db_lock_task is None:
 729             self.logger.debug(
 730                 "Task {} operation={} already locked by another worker".format(
 731                     topic, op_id
 732                 )
 733             )
 734             return False
 735         else:
 736             # Set 'detailed-status' to 'In progress' for VIM/WIM/SDN operations
 737             if self._is_account_type_HA(topic):
 738                 detailed_status = "In progress"
 739                 account_id, op_index = self._get_account_and_op_HA(op_id)
 740                 q_filter = {"_id": account_id}
 741                 update_dict = {
 742                     "_admin.operations.{}.detailed-status".format(
 743                         op_index
 744                     ): detailed_status
 745                 }
 746                 self.db.set_one(
 747                     db_table_name,
 748                     q_filter=q_filter,
 749                     update_dict=update_dict,
 750                     fail_on_empty=False,
 751                 )
 752             return True
 753
 754     def unlock_HA(self, topic, op_type, op_id, operationState, detailed_status):
 755         """
 756         Register a task, done when finished a VIM/WIM/SDN 'create' operation.
 757         :param topic: Can be "vim", "wim", or "sdn"
 758         :param op_type: Operation type, can be "create", "edit", "delete"
 759         :param op_id: Account ID + ':' + Operation Index
 760         :return: nothing
 761         """
 762
 763         # Backward compatibility
 764         if not self._is_account_type_HA(topic) or not op_id:
 765             return
 766
 767         # Get Account ID and Operation Index
 768         account_id, op_index = self._get_account_and_op_HA(op_id)
 769         db_table_name = self.topic2dbtable_dict[topic]
 770
 771         # If this is a 'delete' operation, the account may have been deleted (SUCCESS) or may still exist (FAILED)
 772         # If the account exist, register the HA task.
 773         # Update DB for HA tasks
 774         q_filter = {"_id": account_id}
 775         update_dict = {
 776             "_admin.operations.{}.operationState".format(op_index): operationState,
 777             "_admin.operations.{}.detailed-status".format(op_index): detailed_status,
 778             "_admin.operations.{}.worker".format(op_index): None,
 779             "_admin.current_operation": None,
 780         }
 781         self.db.set_one(
 782             db_table_name,
 783             q_filter=q_filter,
 784             update_dict=update_dict,
 785             fail_on_empty=False,
 786         )
 787         return
 788
 789     async def waitfor_related_HA(self, topic, op_type, op_id=None):
 790         """
 791         Wait for any pending related HA tasks
 792         """
 793
 794         # Backward compatibility
 795         if not (
 796             self._is_service_type_HA(topic) or self._is_account_type_HA(topic)
 797         ) and (op_id is None):
 798             return
 799
 800         # Get DB table name
 801         db_table_name = self.topic2dbtable_dict.get(topic)
 802
 803         # Get instance ID
 804         _id = self._get_instance_id_HA(topic, op_type, op_id)
 805         _filter = {"_id": _id}
 806         db_lcmop = self.db.get_one(db_table_name, _filter, fail_on_empty=False)
 807         if not db_lcmop:
 808             return
 809
 810         # Set DB _filter for querying any related process state
 811         _filter = self._get_waitfor_filter_HA(db_lcmop, topic, op_type, op_id)
 812
 813         # For HA, get list of tasks from DB instead of from dictionary (in-memory) variable.
 814         timeout_wait_for_task = (
 815             3600  # Max time (seconds) to wait for a related task to finish
 816         )
 817         # interval_wait_for_task = 30    #  A too long polling interval slows things down considerably
 818         interval_wait_for_task = 10  # Interval in seconds for polling related tasks
 819         time_left = timeout_wait_for_task
 820         old_num_related_tasks = 0
 821         while True:
 822             # Get related tasks (operations within the same instance as this) which are
 823             # still running (operationState='PROCESSING') and which were started before this task.
 824             # In the case of op_type='ANY', get any related tasks with operationState='PROCESSING', ignore timestamps.
 825             db_waitfor_related_task = self.db.get_list(db_table_name, q_filter=_filter)
 826             new_num_related_tasks = len(db_waitfor_related_task)
 827             # If there are no related tasks, there is nothing to wait for, so return.
 828             if not new_num_related_tasks:
 829                 return
 830             # If number of pending related tasks have changed,
 831             # update the 'detailed-status' field and log the change.
 832             # Do NOT update the 'detailed-status' for SDNC-associated-to-VIM operations ('ANY').
 833             if (op_type != "ANY") and (new_num_related_tasks != old_num_related_tasks):
 834                 step = "Waiting for {} related tasks to be completed.".format(
 835                     new_num_related_tasks
 836                 )
 837                 update_dict = {}
 838                 q_filter = {"_id": _id}
 839                 # NS/NSI
 840                 if self._is_service_type_HA(topic):
 841                     update_dict = {
 842                         "detailed-status": step,
 843                         "queuePosition": new_num_related_tasks,
 844                     }
 845                 # VIM/WIM/SDN
 846                 elif self._is_account_type_HA(topic):
 847                     _, op_index = self._get_account_and_op_HA(op_id)
 848                     update_dict = {
 849                         "_admin.operations.{}.detailed-status".format(op_index): step
 850                     }
 851                 self.logger.debug("Task {} operation={} {}".format(topic, _id, step))
 852                 self.db.set_one(
 853                     db_table_name,
 854                     q_filter=q_filter,
 855                     update_dict=update_dict,
 856                     fail_on_empty=False,
 857                 )
 858                 old_num_related_tasks = new_num_related_tasks
 859             time_left -= interval_wait_for_task
 860             if time_left < 0:
 861                 raise LcmException(
 862                     "Timeout ({}) when waiting for related tasks to be completed".format(
 863                         timeout_wait_for_task
 864                     )
 865                 )
 866             await asyncio.sleep(interval_wait_for_task)
 867
 868         return