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