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