osm_nbi/notifications.py

   1 # Copyright 2020 K Sai Kiran (Tata Elxsi)
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #    http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  12 # implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 __author__ = "K Sai Kiran <saikiran.k@tataelxsi.co.in>"
  17 __date__ = "$28-Apr-2020 23:59:59$"
  18
  19 import asyncio
  20 import aiohttp
  21 from http import HTTPStatus
  22 import json
  23 import logging
  24 import time
  25 from uuid import uuid4
  26
  27
  28 class NotificationException(Exception):
  29     """
  30     Notification Exception
  31     """
  32
  33     def __init__(self, message: str, http_code: int = HTTPStatus.BAD_REQUEST) -> None:
  34         """
  35         Constructor of notification exception
  36         :param message: String text containing exception details.
  37         :param http_code: HTTP status code of exception.
  38         """
  39         self.http_code = http_code
  40         Exception.__init__(self, message)
  41
  42
  43 class NotificationBase:
  44     response_models = None
  45     # Common HTTP payload header for all notifications.
  46     payload_header = {"Content-Type": "application/json", "Accept": "application/json"}
  47
  48     def __init__(self, db) -> None:
  49         """
  50         Constructor of NotificationBase class.
  51         :param db: Database handler.
  52         """
  53         self.db = db
  54         self.logger = logging.getLogger("nbi.notifications")
  55         self.subscriber_collection = None
  56
  57     def get_models(self) -> dict:
  58         """
  59         Returns the SOL005 model of notification class
  60         :param None
  61         :return: dict of SOL005 data model
  62         """
  63         return NotificationBase.response_models
  64
  65     def get_subscribers(self, **kwargs) -> NotificationException:
  66         """
  67         Method should be implemented by all notification subclasses
  68         :param kwargs: any keyword arguments needed for db query.
  69         :return: List of subscribers
  70         """
  71         raise NotificationException(
  72             "Method get_subscribers() is not implemented",
  73             http_code=HTTPStatus.NOT_IMPLEMENTED,
  74         )
  75
  76     @staticmethod
  77     def _get_basic_auth(username: str, password: str) -> tuple:
  78         return aiohttp.BasicAuth(username, password)
  79
  80     def _decrypt_password(
  81         self, hashed: str, salt: str, schema_version: str = "1.1"
  82     ) -> str:
  83         return self.db.decrypt(hashed, schema_version, salt=salt)
  84
  85     def get_payload(self, meta_notification: dict) -> dict:
  86         """
  87         Generates SOL005 compliant payload structure and returns them in dictionary.
  88         :param meta_notification: notification meta data which needs to be formatted as SOL005 compliant
  89         :return: A dictionary which is SOL005 compliant.
  90         """
  91         model_name = meta_notification["notificationType"]
  92         response_models = self.get_models()
  93         if not response_models or not response_models.get(model_name):
  94             raise NotificationException(
  95                 "Response model {} is not defined.".format(model_name),
  96                 HTTPStatus.NOT_IMPLEMENTED,
  97             )
  98         model_keys = response_models[model_name]
  99         payload = dict.fromkeys(model_keys, "N/A")
 100         notification_keys = set(meta_notification.keys())
 101         for model_key in model_keys.intersection(notification_keys):
 102             payload[model_key] = meta_notification[model_key]
 103         self.logger.debug(
 104             "Payload generated for subscriber: {} for {}".format(
 105                 payload["subscriptionId"], payload["notificationType"]
 106             )
 107         )
 108         return payload
 109
 110     async def send_notifications(
 111         self,
 112         subscribers: list,
 113     ):
 114         """
 115         Generate tasks for all notification for an event.
 116         :param subscribers: A list of subscribers who want to be notified for event.
 117         """
 118         notifications = []
 119         for subscriber in subscribers:
 120             # Notify without auth
 121             if not subscriber.get("authentication"):
 122                 notifications.append(
 123                     {
 124                         "headers": self.payload_header,
 125                         "payload": self.get_payload(subscriber),
 126                         "CallbackUri": subscriber["CallbackUri"],
 127                     }
 128                 )
 129             elif subscriber["authentication"]["authType"] == "basic":
 130                 salt = subscriber["subscriptionId"]
 131                 hashed_password = subscriber["authentication"]["paramsBasic"][
 132                     "password"
 133                 ]
 134                 password = self._decrypt_password(hashed_password, salt)
 135                 auth_basic = self._get_basic_auth(
 136                     subscriber["authentication"]["paramsBasic"]["userName"], password
 137                 )
 138                 notifications.append(
 139                     {
 140                         "headers": self.payload_header,
 141                         "payload": self.get_payload(subscriber),
 142                         "auth_basic": auth_basic,
 143                         "CallbackUri": subscriber["CallbackUri"],
 144                     }
 145                 )
 146             # TODO add support for AuthType OAuth and TLS after support is added in subscription.
 147             else:
 148                 self.logger.debug(
 149                     "Subscriber {} can not be notified. {} notification auth type is not implemented".format(
 150                         subscriber["subscriptionId"],
 151                         subscriber["authentication"]["authType"],
 152                     )
 153                 )
 154
 155         if notifications:
 156             tasks = []
 157             async with aiohttp.ClientSession() as session:
 158                 for notification in notifications:
 159                     tasks.append(
 160                         asyncio.ensure_future(
 161                             self.send_notification(session, notification),
 162                         )
 163                     )
 164                 await asyncio.gather(*tasks)
 165
 166     async def send_notification(
 167         self,
 168         session: aiohttp.ClientSession,
 169         notification: dict,
 170         retry_count: int = 5,
 171         timeout: float = 5.0,
 172     ):
 173         """
 174         Performs HTTP Post request to notify subscriber. In case if for any reason notification is not sent successfully
 175         after maximum number of reties, then notification is dropped.
 176         :param session: An aiohttp client session object to maintain http session.
 177         :param notification: A dictionary containing all necessary data to make POST request.
 178         :param retry_count: An integer specifying the maximum number of reties for a notification.
 179         :param timeout: A float representing client timeout of each HTTP request.
 180         """
 181         backoff_delay = 1
 182         while retry_count > 0:
 183             try:
 184                 async with session.post(
 185                     url=notification["CallbackUri"],
 186                     headers=notification["headers"],
 187                     auth=notification.get("auth_basic", None),
 188                     data=json.dumps(notification["payload"]),
 189                     timeout=timeout,
 190                 ) as resp:
 191                     # self.logger.debug("Notification response: {}".format(resp.status))
 192                     if resp.status == HTTPStatus.NO_CONTENT:
 193                         self.logger.debug(
 194                             "Notification sent successfully to subscriber {}".format(
 195                                 notification["payload"]["subscriptionId"]
 196                             )
 197                         )
 198                     else:
 199                         error_text = "Erroneous response code: {}, ".format(resp.status)
 200                         error_text += await resp.text()
 201                         raise NotificationException(error_text)
 202                 return True
 203             except Exception as e:
 204                 error_text = type(e).__name__ + ": " + str(e)
 205                 self.logger.debug(
 206                     "Unable to send notification to subscriber {}. Details: {}".format(
 207                         notification["payload"]["subscriptionId"], error_text
 208                     )
 209                 )
 210                 error_detail = {
 211                     "error": type(e).__name__,
 212                     "error_text": str(e),
 213                     "timestamp": time.time(),
 214                 }
 215                 if "error_details" in notification["payload"].keys():
 216                     notification["payload"]["error_details"].append(error_detail)
 217                 else:
 218                     notification["payload"]["error_details"] = [error_detail]
 219                 retry_count -= 1
 220                 backoff_delay *= 2
 221                 self.logger.debug(
 222                     "Retry Notification for subscriber: {} after backoff delay: {} seconds.".format(
 223                         notification["payload"]["subscriptionId"], backoff_delay
 224                     )
 225                 )
 226                 await asyncio.sleep(backoff_delay)
 227         # Dropping notification
 228         self.logger.debug(
 229             "Notification {} sent failed to subscriber:{}.".format(
 230                 notification["payload"]["notificationType"],
 231                 notification["payload"]["subscriptionId"],
 232             )
 233         )
 234         return False
 235
 236
 237 class NsLcmNotification(NotificationBase):
 238     # SOL005 response model for nslcm notifications
 239     response_models = {
 240         "NsLcmOperationOccurrenceNotification": {
 241             "id",
 242             "nsInstanceId",
 243             "nsLcmOpOccId",
 244             "operation",
 245             "notificationType",
 246             "subscriptionId",
 247             "timestamp",
 248             "notificationStatus",
 249             "operationState",
 250             "isAutomaticInvocation",
 251             "affectedVnf",
 252             "affectedVl",
 253             "affectedVnffg",
 254             "affectedNs",
 255             "affectedSap",
 256             "error",
 257             "_links",
 258         },
 259         "NsIdentifierCreationNotification": {
 260             "notificationType",
 261             "subscriptionId",
 262             "timestamp",
 263             "nsInstanceId",
 264             "_links",
 265         },
 266         "NsIdentifierDeletionNotification": {
 267             "notificationType",
 268             "subscriptionId",
 269             "timestamp",
 270             "nsInstanceId",
 271             "_links",
 272         },
 273         "NsChangeNotification": {
 274             "nsInstanceId",
 275             "nsComponentType",
 276             "nsComponentId",
 277             "lcmOpOccIdImpactngNsComponent",
 278             "lcmOpNameImpactingNsComponent",
 279             "lcmOpOccStatusImpactingNsComponent",
 280             "notificationType",
 281             "subscriptionId",
 282             "timeStamp",
 283             "error",
 284             "_links",
 285         },
 286     }
 287
 288     def __init__(self, db) -> None:
 289         """
 290         Constructor of NsLcmNotification class.
 291         :param db: Database handler.
 292         """
 293         super().__init__(db)
 294         self.subscriber_collection = "mapped_subscriptions"
 295
 296     def get_models(self) -> dict:
 297         """
 298         Returns the SOL005 model of notification class
 299         :param None
 300         :return: dict of SOL005 data model
 301         """
 302         return NsLcmNotification.response_models
 303
 304     @staticmethod
 305     def _format_nslcm_subscribers(subscribers: list, event_details: dict) -> list:
 306         """
 307         Formats the raw event details from kakfa message and subscriber details.
 308         :param subscribers: A list of subscribers whom the event needs to be notified.
 309         :param event_details: A dict containing all meta data of event.
 310         :return:
 311         """
 312         notification_id = str(uuid4())
 313         event_timestamp = event_details["params"]["startTime"]
 314         resource_links = event_details["params"]["links"]
 315         event_operation = event_details["command"]
 316         for key in ["_admin", "_id", "id", "links"]:
 317             event_details["params"].pop(key, None)
 318         for subscriber in subscribers:
 319             subscriber["id"] = notification_id
 320             subscriber["timestamp"] = event_timestamp
 321             subscriber["_links"] = resource_links
 322             subscriber["subscriptionId"] = subscriber["reference"]
 323             subscriber["operation"] = event_operation
 324             del subscriber["reference"]
 325             del subscriber["_id"]
 326             subscriber.update(event_details["params"])
 327         return subscribers
 328
 329     def get_subscribers(
 330         self,
 331         nsd_id: str,
 332         ns_instance_id: str,
 333         command: str,
 334         op_state: str,
 335         event_details: dict,
 336     ) -> list:
 337         """
 338         Queries database and returns list of subscribers.
 339         :param nsd_id: NSD id of an NS whose lifecycle has changed. (scaled, terminated. etc)
 340         :param ns_instance_id: NS instance id an NS whose lifecycle has changed.
 341         :param command: the command for event.
 342         :param op_state: the operation state of NS.
 343         :param event_details: dict containing raw data of event occured.
 344         :return: List of interested subscribers for occurred event.
 345         """
 346         notification_type = [
 347             "NsLcmOperationOccurrenceNotification",
 348             "NsChangeNotification",
 349             "NsIdentifierCreationNotification",
 350             "NsIdentifierDeletionNotification",
 351         ]
 352         filter_q = {
 353             "identifier": [nsd_id, ns_instance_id],
 354             "operationStates": ["ANY"],
 355             "operationTypes": ["ANY"],
 356             "notificationType": notification_type,
 357         }
 358         if op_state:
 359             filter_q["operationStates"].append(op_state)
 360         if command:
 361             filter_q["operationTypes"].append(command)
 362         # self.logger.debug("Db query is: {}".format(filter_q))
 363         subscribers = []
 364         try:
 365             subscribers = self.db.get_list(self.subscriber_collection, filter_q)
 366             subscribers = self._format_nslcm_subscribers(subscribers, event_details)
 367         except Exception as e:
 368             error_text = type(e).__name__ + ": " + str(e)
 369             self.logger.debug("Error getting nslcm subscribers: {}".format(error_text))
 370         finally:
 371             return subscribers
 372
 373
 374 class VnfLcmNotification(NotificationBase):
 375     # SOL003 response model for vnflcm notifications
 376     response_models = {
 377         "VnfLcmOperationOccurrenceNotification": {
 378             "id",
 379             "notificationType",
 380             "subscriptionId",
 381             "timeStamp",
 382             "notificationStatus",
 383             "operationState",
 384             "vnfInstanceId",
 385             "operation",
 386             "isAutomaticInvocation",
 387             "vnfLcmOpOccId",
 388             "affectedVnfcs",
 389             "affectedVirtualLinks",
 390             "affectedExtLinkPorts",
 391             "affectedVirtualStorages",
 392             "changedInfo",
 393             "changedExtConnectivity",
 394             "modificationsTriggeredByVnfPkgChange",
 395             "error",
 396             "_links",
 397         },
 398         "VnfIdentifierCreationNotification": {
 399             "id",
 400             "notificationType",
 401             "subscriptionId",
 402             "timeStamp",
 403             "vnfInstanceId",
 404             "_links",
 405         },
 406         "VnfIdentifierDeletionNotification": {
 407             "id",
 408             "notificationType",
 409             "subscriptionId",
 410             "timeStamp",
 411             "vnfInstanceId",
 412             "_links",
 413         },
 414     }
 415
 416     def __init__(self, db) -> None:
 417         """
 418         Constructor of VnfLcmNotification class.
 419         :param db: Database handler.
 420         """
 421         super().__init__(db)
 422         self.subscriber_collection = "mapped_subscriptions"
 423
 424     def get_models(self) -> dict:
 425         """
 426         Returns the SOL003 model of notification class
 427         :param None
 428         :return: dict of SOL003 data model
 429         """
 430         return self.response_models
 431
 432     def _format_vnflcm_subscribers(
 433         self, subscribers: list, event_details: dict
 434     ) -> list:
 435         """
 436         Formats the raw event details from kafka message and subscriber details.
 437         :param subscribers: A list of subscribers whom the event needs to be notified.
 438         :param event_details: A dict containing all meta data of event.
 439         :return:
 440         """
 441         notification_id = str(uuid4())
 442         event_timestamp = time.time()
 443         event_operation = event_details["command"]
 444         for subscriber in subscribers:
 445             subscriber["id"] = notification_id
 446             subscriber["timeStamp"] = event_timestamp
 447             subscriber["subscriptionId"] = subscriber["reference"]
 448             subscriber["operation"] = event_operation
 449             del subscriber["reference"]
 450             del subscriber["_id"]
 451             subscriber.update(event_details["params"])
 452         return subscribers
 453
 454     def get_subscribers(
 455         self,
 456         vnfd_id: str,
 457         vnf_instance_id: str,
 458         command: str,
 459         op_state: str,
 460         event_details: dict,
 461     ) -> list:
 462         """
 463         Queries database and returns list of subscribers.
 464         :param vnfd_id: Vnfd id of a VNF whose lifecycle has changed. (instantiated, scaled, terminated. etc)
 465         :param vnf_instance_id: Vnf instance id of a VNF whose lifecycle has changed.
 466         :param command: the command for event.
 467         :param op_state: the operation state of VNF.
 468         :param event_details: dict containing raw data of event occurred.
 469         :return: List of interested subscribers for occurred event.
 470         """
 471         notification_type = [
 472             "VnfIdentifierCreationNotification",
 473             "VnfLcmOperationOccurrenceNotification",
 474             "VnfIdentifierDeletionNotification",
 475         ]
 476         filter_q = {
 477             "identifier": [vnfd_id, vnf_instance_id],
 478             "operationStates": ["ANY"],
 479             "operationTypes": ["ANY"],
 480             "notificationType": notification_type,
 481         }
 482         if op_state:
 483             filter_q["operationStates"].append(op_state)
 484         if command:
 485             filter_q["operationTypes"].append(command)
 486         subscribers = []
 487         try:
 488             subscribers = self.db.get_list(self.subscriber_collection, filter_q)
 489             subscribers = self._format_vnflcm_subscribers(subscribers, event_details)
 490         except Exception as e:
 491             error_text = type(e).__name__ + ": " + str(e)
 492             self.logger.debug("Error getting vnflcm subscribers: {}".format(error_text))
 493         finally:
 494             return subscribers
 495
 496
 497 class NsdNotification(NotificationBase):
 498     def __init__(self, db):
 499         """
 500         Constructor of the class
 501         """
 502         super().__init__(db)
 503         # TODO will update this once support is there from subscription
 504         self.response_models = {}
 505         self.subscriber_collection = None
 506
 507
 508 class VnfdNotification(NotificationBase):
 509     def __init__(self, db):
 510         """
 511         Constructor of the class
 512         """
 513         super().__init__(db)
 514         # TODO will update this once support is there from subscription
 515         self.response_models = {}
 516         self.subscriber_collection = None