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