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