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