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