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