osm_lcm/lcm_utils.py

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