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