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 from collections import OrderedDict
  21 from time import time
  22 # from osm_common.dbbase import DbException
  23
  24 __author__ = "Alfonso Tierno"
  25
  26
  27 class LcmException(Exception):
  28     pass
  29
  30
  31 class LcmExceptionNoMgmtIP(LcmException):
  32     pass
  33
  34
  35 class LcmExceptionExit(LcmException):
  36     pass
  37
  38
  39 def versiontuple(v):
  40     """utility for compare dot separate versions. Fills with zeros to proper number comparison
  41     package version will be something like 4.0.1.post11+gb3f024d.dirty-1. Where 4.0.1 is the git tag, postXX is the
  42     number of commits from this tag, and +XXXXXXX is the git commit short id. Total length is 16 with until 999 commits
  43     """
  44     filled = []
  45     for point in v.split("."):
  46         point, _, _ = point.partition("+")
  47         point, _, _ = point.partition("-")
  48         filled.append(point.zfill(20))
  49     return tuple(filled)
  50
  51
  52 def deep_get(target_dict, key_list):
  53     """
  54     Get a value from target_dict entering in the nested keys. If keys does not exist, it returns None
  55     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
  56     :param target_dict: dictionary to be read
  57     :param key_list: list of keys to read from  target_dict
  58     :return: The wanted value if exist, None otherwise
  59     """
  60     for key in key_list:
  61         if not isinstance(target_dict, dict) or key not in target_dict:
  62             return None
  63         target_dict = target_dict[key]
  64     return target_dict
  65
  66
  67 # LcmBase must be listed before TaskRegistry, as it is a dependency.
  68 class LcmBase:
  69
  70     def __init__(self, db, msg, fs, logger):
  71         """
  72
  73         :param db: database connection
  74         """
  75         self.db = db
  76         self.msg = msg
  77         self.fs = fs
  78         self.logger = logger
  79
  80     def update_db_2(self, item, _id, _desc):
  81         """
  82         Updates database with _desc information. If success _desc is cleared
  83         :param item:
  84         :param _id:
  85         :param _desc: dictionary with the content to update. Keys are dot separated keys for
  86         :return: None. Exception is raised on error
  87         """
  88         if not _desc:
  89             return
  90         now = time()
  91         _desc["_admin.modified"] = now
  92         self.db.set_one(item, {"_id": _id}, _desc)
  93         _desc.clear()
  94         # except DbException as e:
  95         #     self.logger.error("Updating {} _id={} with '{}'. Error: {}".format(item, _id, _desc, e))
  96
  97
  98 class TaskRegistry(LcmBase):
  99     """
 100     Implements a registry of task needed for later cancelation, look for related tasks that must be completed before
 101     etc. It stores a four level dict
 102     First level is the topic, ns, vim_account, sdn
 103     Second level is the _id
 104     Third level is the operation id
 105     Fourth level is a descriptive name, the value is the task class
 106
 107     The HA (High-Availability) methods are used when more than one LCM instance is running.
 108     To register the current task in the external DB, use LcmBase as base class, to be able
 109     to reuse LcmBase.update_db_2()
 110     The DB registry uses the following fields to distinguish a task:
 111     - op_type: operation type ("nslcmops" or "nsilcmops")
 112     - op_id:   operation ID
 113     - worker:  the worker ID for this process
 114     """
 115
 116     # NS/NSI: "services" VIM/WIM/SDN: "accounts"
 117     topic_service_list = ['ns', 'nsi']
 118     topic_account_list = ['vim', 'wim', 'sdn', 'k8scluster', 'k8srepo']
 119
 120     # Map topic to InstanceID
 121     topic2instid_dict = {
 122         'ns': 'nsInstanceId',
 123         'nsi': 'netsliceInstanceId'}
 124
 125     # Map topic to DB table name
 126     topic2dbtable_dict = {
 127         'ns': 'nslcmops',
 128         'nsi': 'nsilcmops',
 129         'vim': 'vim_accounts',
 130         'wim': 'wim_accounts',
 131         'sdn': 'sdns',
 132         'k8scluster': 'k8sclusters',
 133         'k8srepo': 'k8srepos'}
 134
 135     def __init__(self, worker_id=None, db=None, logger=None):
 136         self.task_registry = {
 137             "ns": {},
 138             "nsi": {},
 139             "vim_account": {},
 140             "wim_account": {},
 141             "sdn": {},
 142             "k8scluster": {},
 143             "k8srepo": {},
 144         }
 145         self.worker_id = worker_id
 146         self.db = db
 147         self.logger = logger
 148
 149     def register(self, topic, _id, op_id, task_name, task):
 150         """
 151         Register a new task
 152         :param topic: Can be "ns", "nsi", "vim_account", "sdn"
 153         :param _id: _id of the related item
 154         :param op_id: id of the operation of the related item
 155         :param task_name: Task descriptive name, as create, instantiate, terminate. Must be unique in this op_id
 156         :param task: Task class
 157         :return: none
 158         """
 159         if _id not in self.task_registry[topic]:
 160             self.task_registry[topic][_id] = OrderedDict()
 161         if op_id not in self.task_registry[topic][_id]:
 162             self.task_registry[topic][_id][op_id] = {task_name: task}
 163         else:
 164             self.task_registry[topic][_id][op_id][task_name] = task
 165         # print("registering task", topic, _id, op_id, task_name, task)
 166
 167     def remove(self, topic, _id, op_id, task_name=None):
 168         """
 169         When task is ended, it should be removed. It ignores missing tasks. It also removes tasks done with this _id
 170         :param topic: Can be "ns", "nsi", "vim_account", "sdn"
 171         :param _id: _id of the related item
 172         :param op_id: id of the operation of the related item
 173         :param task_name: Task descriptive name. If none it deletes all tasks with same _id and op_id
 174         :return: None
 175         """
 176         if not self.task_registry[topic].get(_id):
 177             return
 178         if not task_name:
 179             self.task_registry[topic][_id].pop(op_id, None)
 180         elif self.task_registry[topic][_id].get(op_id):
 181             self.task_registry[topic][_id][op_id].pop(task_name, None)
 182
 183         # delete done tasks
 184         for op_id_ in list(self.task_registry[topic][_id]):
 185             for name, task in self.task_registry[topic][_id][op_id_].items():
 186                 if not task.done():
 187                     break
 188             else:
 189                 del self.task_registry[topic][_id][op_id_]
 190         if not self.task_registry[topic][_id]:
 191             del self.task_registry[topic][_id]
 192
 193     def lookfor_related(self, topic, _id, my_op_id=None):
 194         task_list = []
 195         task_name_list = []
 196         if _id not in self.task_registry[topic]:
 197             return "", task_name_list
 198         for op_id in reversed(self.task_registry[topic][_id]):
 199             if my_op_id:
 200                 if my_op_id == op_id:
 201                     my_op_id = None  # so that the next task is taken
 202                 continue
 203
 204             for task_name, task in self.task_registry[topic][_id][op_id].items():
 205                 if not task.done():
 206                     task_list.append(task)
 207                     task_name_list.append(task_name)
 208             break
 209         return ", ".join(task_name_list), task_list
 210
 211     def cancel(self, topic, _id, target_op_id=None, target_task_name=None):
 212         """
 213         Cancel all active tasks of a concrete ns, nsi, vim_account, sdn identified for _id. If op_id is supplied only
 214         this is cancelled, and the same with task_name
 215         """
 216         if not self.task_registry[topic].get(_id):
 217             return
 218         for op_id in reversed(self.task_registry[topic][_id]):
 219             if target_op_id and target_op_id != op_id:
 220                 continue
 221             for task_name, task in self.task_registry[topic][_id][op_id].items():
 222                 if target_task_name and target_task_name != task_name:
 223                     continue
 224                 # result =
 225                 task.cancel()
 226                 # if result:
 227                 #     self.logger.debug("{} _id={} order_id={} task={} cancelled".format(topic, _id, op_id, task_name))
 228
 229     # Is topic NS/NSI?
 230     def _is_service_type_HA(self, topic):
 231         return topic in self.topic_service_list
 232
 233     # Is topic VIM/WIM/SDN?
 234     def _is_account_type_HA(self, topic):
 235         return topic in self.topic_account_list
 236
 237     # Input: op_id, example: 'abc123def:3' Output: account_id='abc123def', op_index=3
 238     def _get_account_and_op_HA(self, op_id):
 239         if not op_id:
 240             return (None, None)
 241         account_id, _, op_index = op_id.rpartition(':')
 242         if not account_id:
 243             return (None, None)
 244         if not op_index.isdigit():
 245             return (None, None)
 246         return account_id, op_index
 247
 248     # Get '_id' for any topic and operation
 249     def _get_instance_id_HA(self, topic, op_type, op_id):
 250         _id = None
 251         # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
 252         if op_type == 'ANY':
 253             _id = op_id
 254         # NS/NSI: Use op_id as '_id'
 255         elif self._is_service_type_HA(topic):
 256             _id = op_id
 257         # VIM/SDN/WIM/K8SCLUSTER: Split op_id to get Account ID and Operation Index, use Account ID as '_id'
 258         elif self._is_account_type_HA(topic):
 259             _id, _ = self._get_account_and_op_HA(op_id)
 260         return _id
 261
 262     # Set DB _filter for querying any related process state
 263     def _get_waitfor_filter_HA(self, db_lcmop, topic, op_type, op_id):
 264         _filter = {}
 265         # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
 266         # In this special case, the timestamp is ignored
 267         if op_type == 'ANY':
 268             _filter = {'operationState': 'PROCESSING'}
 269         # Otherwise, get 'startTime' timestamp for this operation
 270         else:
 271             # NS/NSI
 272             if self._is_service_type_HA(topic):
 273                 now = time()
 274                 starttime_this_op = db_lcmop.get("startTime")
 275                 instance_id_label = self.topic2instid_dict.get(topic)
 276                 instance_id = db_lcmop.get(instance_id_label)
 277                 _filter = {instance_id_label: instance_id,
 278                            'operationState': 'PROCESSING',
 279                            'startTime.lt': starttime_this_op,
 280                            "_admin.modified.gt": now - 2*3600,  # ignore if tow hours of inactivity
 281                            }
 282             # VIM/WIM/SDN/K8scluster
 283             elif self._is_account_type_HA(topic):
 284                 _, op_index = self._get_account_and_op_HA(op_id)
 285                 _ops = db_lcmop['_admin']['operations']
 286                 _this_op = _ops[int(op_index)]
 287                 starttime_this_op = _this_op.get('startTime', None)
 288                 _filter = {'operationState': 'PROCESSING',
 289                            'startTime.lt': starttime_this_op}
 290         return _filter
 291
 292     # Get DB params for any topic and operation
 293     def _get_dbparams_for_lock_HA(self, topic, op_type, op_id):
 294         q_filter = {}
 295         update_dict = {}
 296         # NS/NSI
 297         if self._is_service_type_HA(topic):
 298             q_filter = {'_id': op_id, '_admin.worker': None}
 299             update_dict = {'_admin.worker': self.worker_id}
 300         # VIM/WIM/SDN
 301         elif self._is_account_type_HA(topic):
 302             account_id, op_index = self._get_account_and_op_HA(op_id)
 303             if not account_id:
 304                 return None, None
 305             if op_type == 'create':
 306                 # Creating a VIM/WIM/SDN account implies setting '_admin.current_operation' = 0
 307                 op_index = 0
 308             q_filter = {'_id': account_id, "_admin.operations.{}.worker".format(op_index): None}
 309             update_dict = {'_admin.operations.{}.worker'.format(op_index): self.worker_id,
 310                            '_admin.current_operation': op_index}
 311         return q_filter, update_dict
 312
 313     def lock_HA(self, topic, op_type, op_id):
 314         """
 315         Lock a task, if possible, to indicate to the HA system that
 316         the task will be executed in this LCM instance.
 317         :param topic: Can be "ns", "nsi", "vim", "wim", or "sdn"
 318         :param op_type: Operation type, can be "nslcmops", "nsilcmops", "create", "edit", "delete"
 319         :param op_id: NS, NSI: Operation ID  VIM,WIM,SDN: Account ID + ':' + Operation Index
 320         :return:
 321         True=lock was successful => execute the task (not registered by any other LCM instance)
 322         False=lock failed => do NOT execute the task (already registered by another LCM instance)
 323
 324         HA tasks and backward compatibility:
 325         If topic is "account type" (VIM/WIM/SDN) and op_id is None, 'op_id' was not provided by NBI.
 326         This means that the running NBI instance does not support HA.
 327         In such a case this method should always return True, to always execute
 328         the task in this instance of LCM, without querying the DB.
 329         """
 330
 331         # Backward compatibility for VIM/WIM/SDN/k8scluster without op_id
 332         if self._is_account_type_HA(topic) and op_id is None:
 333             return True
 334
 335         # Try to lock this task
 336         db_table_name = self.topic2dbtable_dict.get(topic)
 337         q_filter, update_dict = self._get_dbparams_for_lock_HA(topic, op_type, op_id)
 338         db_lock_task = self.db.set_one(db_table_name,
 339                                        q_filter=q_filter,
 340                                        update_dict=update_dict,
 341                                        fail_on_empty=False)
 342         if db_lock_task is None:
 343             self.logger.debug("Task {} operation={} already locked by another worker".format(topic, op_id))
 344             return False
 345         else:
 346             # Set 'detailed-status' to 'In progress' for VIM/WIM/SDN operations
 347             if self._is_account_type_HA(topic):
 348                 detailed_status = 'In progress'
 349                 account_id, op_index = self._get_account_and_op_HA(op_id)
 350                 q_filter = {'_id': account_id}
 351                 update_dict = {'_admin.operations.{}.detailed-status'.format(op_index): detailed_status}
 352                 self.db.set_one(db_table_name,
 353                                 q_filter=q_filter,
 354                                 update_dict=update_dict,
 355                                 fail_on_empty=False)
 356             return True
 357
 358     def register_HA(self, topic, op_type, op_id, operationState, detailed_status):
 359         """
 360         Register a task, done when finished a VIM/WIM/SDN 'create' operation.
 361         :param topic: Can be "vim", "wim", or "sdn"
 362         :param op_type: Operation type, can be "create", "edit", "delete"
 363         :param op_id: Account ID + ':' + Operation Index
 364         :return: nothing
 365         """
 366
 367         # Backward compatibility
 368         if not self._is_account_type_HA(topic) or (self._is_account_type_HA(topic) and op_id is None):
 369             return
 370
 371         # Get Account ID and Operation Index
 372         account_id, op_index = self._get_account_and_op_HA(op_id)
 373         db_table_name = self.topic2dbtable_dict.get(topic)
 374
 375         # If this is a 'delete' operation, the account may have been deleted (SUCCESS) or may still exist (FAILED)
 376         # If the account exist, register the HA task.
 377         # Update DB for HA tasks
 378         q_filter = {'_id': account_id}
 379         update_dict = {'_admin.operations.{}.operationState'.format(op_index): operationState,
 380                        '_admin.operations.{}.detailed-status'.format(op_index): detailed_status}
 381         self.db.set_one(db_table_name,
 382                         q_filter=q_filter,
 383                         update_dict=update_dict,
 384                         fail_on_empty=False)
 385         return
 386
 387     async def waitfor_related_HA(self, topic, op_type, op_id=None):
 388         """
 389         Wait for any pending related HA tasks
 390         """
 391
 392         # Backward compatibility
 393         if not (self._is_service_type_HA(topic) or self._is_account_type_HA(topic)) and (op_id is None):
 394             return
 395
 396         # Get DB table name
 397         db_table_name = self.topic2dbtable_dict.get(topic)
 398
 399         # Get instance ID
 400         _id = self._get_instance_id_HA(topic, op_type, op_id)
 401         _filter = {"_id": _id}
 402         db_lcmop = self.db.get_one(db_table_name,
 403                                    _filter,
 404                                    fail_on_empty=False)
 405         if not db_lcmop:
 406             return
 407
 408         # Set DB _filter for querying any related process state
 409         _filter = self._get_waitfor_filter_HA(db_lcmop, topic, op_type, op_id)
 410
 411         # For HA, get list of tasks from DB instead of from dictionary (in-memory) variable.
 412         timeout_wait_for_task = 3600   # Max time (seconds) to wait for a related task to finish
 413         # interval_wait_for_task = 30    #  A too long polling interval slows things down considerably
 414         interval_wait_for_task = 10       # Interval in seconds for polling related tasks
 415         time_left = timeout_wait_for_task
 416         old_num_related_tasks = 0
 417         while True:
 418             # Get related tasks (operations within the same instance as this) which are
 419             # still running (operationState='PROCESSING') and which were started before this task.
 420             # In the case of op_type='ANY', get any related tasks with operationState='PROCESSING', ignore timestamps.
 421             db_waitfor_related_task = self.db.get_list(db_table_name,
 422                                                        q_filter=_filter)
 423             new_num_related_tasks = len(db_waitfor_related_task)
 424             # If there are no related tasks, there is nothing to wait for, so return.
 425             if not new_num_related_tasks:
 426                 return
 427             # If number of pending related tasks have changed,
 428             # update the 'detailed-status' field and log the change.
 429             # Do NOT update the 'detailed-status' for SDNC-associated-to-VIM operations ('ANY').
 430             if (op_type != 'ANY') and (new_num_related_tasks != old_num_related_tasks):
 431                 step = "Waiting for {} related tasks to be completed.".format(new_num_related_tasks)
 432                 update_dict = {}
 433                 q_filter = {'_id': _id}
 434                 # NS/NSI
 435                 if self._is_service_type_HA(topic):
 436                     update_dict = {'detailed-status': step}
 437                 # VIM/WIM/SDN
 438                 elif self._is_account_type_HA(topic):
 439                     _, op_index = self._get_account_and_op_HA(op_id)
 440                     update_dict = {'_admin.operations.{}.detailed-status'.format(op_index): step}
 441                 self.logger.debug("Task {} operation={} {}".format(topic, _id, step))
 442                 self.db.set_one(db_table_name,
 443                                 q_filter=q_filter,
 444                                 update_dict=update_dict,
 445                                 fail_on_empty=False)
 446                 old_num_related_tasks = new_num_related_tasks
 447             time_left -= interval_wait_for_task
 448             if time_left < 0:
 449                 raise LcmException(
 450                     "Timeout ({}) when waiting for related tasks to be completed".format(
 451                         timeout_wait_for_task))
 452             await asyncio.sleep(interval_wait_for_task)
 453
 454         return