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