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