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, subscribers: list, loop: asyncio.AbstractEventLoop = None
 112     ):
 113         """
 114         Generate tasks for all notification for an event.
 115         :param subscribers: A list of subscribers who want to be notified for event.
 116         :param loop: Event loop object.
 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(loop=loop) as session:
 158                 for notification in notifications:
 159                     tasks.append(
 160                         asyncio.ensure_future(
 161                             self.send_notification(session, notification, loop=loop),
 162                             loop=loop,
 163                         )
 164                     )
 165                 await asyncio.gather(*tasks, loop=loop)
 166
 167     async def send_notification(
 168         self,
 169         session: aiohttp.ClientSession,
 170         notification: dict,
 171         loop: asyncio.AbstractEventLoop = None,
 172         retry_count: int = 5,
 173         timeout: float = 5.0,
 174     ):
 175         """
 176         Performs HTTP Post request to notify subscriber. In case if for any reason notification is not sent successfully
 177         after maximum number of reties, then notification is dropped.
 178         :param session: An aiohttp client session object to maintain http session.
 179         :param notification: A dictionary containing all necessary data to make POST request.
 180         :param loop: Event loop object.
 181         :param retry_count: An integer specifying the maximum number of reties for a notification.
 182         :param timeout: A float representing client timeout of each HTTP request.
 183         """
 184         backoff_delay = 1
 185         while retry_count > 0:
 186             try:
 187                 async with session.post(
 188                     url=notification["CallbackUri"],
 189                     headers=notification["headers"],
 190                     auth=notification.get("auth_basic", None),
 191                     data=json.dumps(notification["payload"]),
 192                     timeout=timeout,
 193                 ) as resp:
 194                     # self.logger.debug("Notification response: {}".format(resp.status))
 195                     if resp.status == HTTPStatus.NO_CONTENT:
 196                         self.logger.debug(
 197                             "Notification sent successfully to subscriber {}".format(
 198                                 notification["payload"]["subscriptionId"]
 199                             )
 200                         )
 201                     else:
 202                         error_text = "Erroneous response code: {}, ".format(resp.status)
 203                         error_text += await resp.text()
 204                         raise NotificationException(error_text)
 205                 return True
 206             except Exception as e:
 207                 error_text = type(e).__name__ + ": " + str(e)
 208                 self.logger.debug(
 209                     "Unable to send notification to subscriber {}. Details: {}".format(
 210                         notification["payload"]["subscriptionId"], error_text
 211                     )
 212                 )
 213                 error_detail = {
 214                     "error": type(e).__name__,
 215                     "error_text": str(e),
 216                     "timestamp": time.time(),
 217                 }
 218                 if "error_details" in notification["payload"].keys():
 219                     notification["payload"]["error_details"].append(error_detail)
 220                 else:
 221                     notification["payload"]["error_details"] = [error_detail]
 222                 retry_count -= 1
 223                 backoff_delay *= 2
 224                 self.logger.debug(
 225                     "Retry Notification for subscriber: {} after backoff delay: {} seconds.".format(
 226                         notification["payload"]["subscriptionId"], backoff_delay
 227                     )
 228                 )
 229                 await asyncio.sleep(backoff_delay, loop=loop)
 230         # Dropping notification
 231         self.logger.debug(
 232             "Notification {} sent failed to subscriber:{}.".format(
 233                 notification["payload"]["notificationType"],
 234                 notification["payload"]["subscriptionId"],
 235             )
 236         )
 237         return False
 238
 239
 240 class NsLcmNotification(NotificationBase):
 241     # SOL005 response model for nslcm notifications
 242     response_models = {
 243         "NsLcmOperationOccurrenceNotification": {
 244             "id",
 245             "nsInstanceId",
 246             "nsLcmOpOccId",
 247             "operation",
 248             "notificationType",
 249             "subscriptionId",
 250             "timestamp",
 251             "notificationStatus",
 252             "operationState",
 253             "isAutomaticInvocation",
 254             "affectedVnf",
 255             "affectedVl",
 256             "affectedVnffg",
 257             "affectedNs",
 258             "affectedSap",
 259             "error",
 260             "_links",
 261         },
 262         "NsIdentifierCreationNotification": {
 263             "notificationType",
 264             "subscriptionId",
 265             "timestamp",
 266             "nsInstanceId",
 267             "_links",
 268         },
 269         "NsIdentifierDeletionNotification": {
 270             "notificationType",
 271             "subscriptionId",
 272             "timestamp",
 273             "nsInstanceId",
 274             "_links",
 275         },
 276         "NsChangeNotification": {
 277             "nsInstanceId",
 278             "nsComponentType",
 279             "nsComponentId",
 280             "lcmOpOccIdImpactngNsComponent",
 281             "lcmOpNameImpactingNsComponent",
 282             "lcmOpOccStatusImpactingNsComponent",
 283             "notificationType",
 284             "subscriptionId",
 285             "timeStamp",
 286             "error",
 287             "_links",
 288         },
 289     }
 290
 291     def __init__(self, db) -> None:
 292         """
 293         Constructor of NsLcmNotification class.
 294         :param db: Database handler.
 295         """
 296         super().__init__(db)
 297         self.subscriber_collection = "mapped_subscriptions"
 298
 299     def get_models(self) -> dict:
 300         """
 301         Returns the SOL005 model of notification class
 302         :param None
 303         :return: dict of SOL005 data model
 304         """
 305         return NsLcmNotification.response_models
 306
 307     @staticmethod
 308     def _format_nslcm_subscribers(subscribers: list, event_details: dict) -> list:
 309         """
 310         Formats the raw event details from kakfa message and subscriber details.
 311         :param subscribers: A list of subscribers whom the event needs to be notified.
 312         :param event_details: A dict containing all meta data of event.
 313         :return:
 314         """
 315         notification_id = str(uuid4())
 316         event_timestamp = event_details["params"]["startTime"]
 317         resource_links = event_details["params"]["links"]
 318         event_operation = event_details["command"]
 319         for key in ["_admin", "_id", "id", "links"]:
 320             event_details["params"].pop(key, None)
 321         for subscriber in subscribers:
 322             subscriber["id"] = notification_id
 323             subscriber["timestamp"] = event_timestamp
 324             subscriber["_links"] = resource_links
 325             subscriber["subscriptionId"] = subscriber["reference"]
 326             subscriber["operation"] = event_operation
 327             del subscriber["reference"]
 328             del subscriber["_id"]
 329             subscriber.update(event_details["params"])
 330         return subscribers
 331
 332     def get_subscribers(
 333         self,
 334         nsd_id: str,
 335         ns_instance_id: str,
 336         command: str,
 337         op_state: str,
 338         event_details: dict,
 339     ) -> list:
 340         """
 341         Queries database and returns list of subscribers.
 342         :param nsd_id: NSD id of an NS whose lifecycle has changed. (scaled, terminated. etc)
 343         :param ns_instance_id: NS instance id an NS whose lifecycle has changed.
 344         :param command: the command for event.
 345         :param op_state: the operation state of NS.
 346         :param event_details: dict containing raw data of event occured.
 347         :return: List of interested subscribers for occurred event.
 348         """
 349         notification_type = [
 350             "NsLcmOperationOccurrenceNotification",
 351             "NsChangeNotification",
 352             "NsIdentifierCreationNotification",
 353             "NsIdentifierDeletionNotification",
 354         ]
 355         filter_q = {
 356             "identifier": [nsd_id, ns_instance_id],
 357             "operationStates": ["ANY"],
 358             "operationTypes": ["ANY"],
 359             "notificationType": notification_type,
 360         }
 361         if op_state:
 362             filter_q["operationStates"].append(op_state)
 363         if command:
 364             filter_q["operationTypes"].append(command)
 365         # self.logger.debug("Db query is: {}".format(filter_q))
 366         subscribers = []
 367         try:
 368             subscribers = self.db.get_list(self.subscriber_collection, filter_q)
 369             subscribers = self._format_nslcm_subscribers(subscribers, event_details)
 370         except Exception as e:
 371             error_text = type(e).__name__ + ": " + str(e)
 372             self.logger.debug("Error getting nslcm subscribers: {}".format(error_text))
 373         finally:
 374             return subscribers
 375
 376
 377 class VnfLcmNotification(NotificationBase):
 378     # SOL003 response model for vnflcm notifications
 379     response_models = {
 380         "VnfLcmOperationOccurrenceNotification": {
 381             "id",
 382             "notificationType",
 383             "subscriptionId",
 384             "timeStamp",
 385             "notificationStatus",
 386             "operationState",
 387             "vnfInstanceId",
 388             "operation",
 389             "isAutomaticInvocation",
 390             "vnfLcmOpOccId",
 391             "affectedVnfcs",
 392             "affectedVirtualLinks",
 393             "affectedExtLinkPorts",
 394             "affectedVirtualStorages",
 395             "changedInfo",
 396             "changedExtConnectivity",
 397             "modificationsTriggeredByVnfPkgChange",
 398             "error",
 399             "_links",
 400         },
 401         "VnfIdentifierCreationNotification": {
 402             "id",
 403             "notificationType",
 404             "subscriptionId",
 405             "timeStamp",
 406             "vnfInstanceId",
 407             "_links",
 408         },
 409         "VnfIdentifierDeletionNotification": {
 410             "id",
 411             "notificationType",
 412             "subscriptionId",
 413             "timeStamp",
 414             "vnfInstanceId",
 415             "_links",
 416         },
 417     }
 418
 419     def __init__(self, db) -> None:
 420         """
 421         Constructor of VnfLcmNotification class.
 422         :param db: Database handler.
 423         """
 424         super().__init__(db)
 425         self.subscriber_collection = "mapped_subscriptions"
 426
 427     def get_models(self) -> dict:
 428         """
 429         Returns the SOL003 model of notification class
 430         :param None
 431         :return: dict of SOL003 data model
 432         """
 433         return self.response_models
 434
 435     def _format_vnflcm_subscribers(
 436         self, subscribers: list, event_details: dict
 437     ) -> list:
 438         """
 439         Formats the raw event details from kafka message and subscriber details.
 440         :param subscribers: A list of subscribers whom the event needs to be notified.
 441         :param event_details: A dict containing all meta data of event.
 442         :return:
 443         """
 444         notification_id = str(uuid4())
 445         event_timestamp = time.time()
 446         event_operation = event_details["command"]
 447         for subscriber in subscribers:
 448             subscriber["id"] = notification_id
 449             subscriber["timeStamp"] = event_timestamp
 450             subscriber["subscriptionId"] = subscriber["reference"]
 451             subscriber["operation"] = event_operation
 452             del subscriber["reference"]
 453             del subscriber["_id"]
 454             subscriber.update(event_details["params"])
 455         return subscribers
 456
 457     def get_subscribers(
 458         self,
 459         vnfd_id: str,
 460         vnf_instance_id: str,
 461         command: str,
 462         op_state: str,
 463         event_details: dict,
 464     ) -> list:
 465         """
 466         Queries database and returns list of subscribers.
 467         :param vnfd_id: Vnfd id of a VNF whose lifecycle has changed. (instantiated, scaled, terminated. etc)
 468         :param vnf_instance_id: Vnf instance id of a VNF whose lifecycle has changed.
 469         :param command: the command for event.
 470         :param op_state: the operation state of VNF.
 471         :param event_details: dict containing raw data of event occurred.
 472         :return: List of interested subscribers for occurred event.
 473         """
 474         notification_type = [
 475             "VnfIdentifierCreationNotification",
 476             "VnfLcmOperationOccurrenceNotification",
 477             "VnfIdentifierDeletionNotification",
 478         ]
 479         filter_q = {
 480             "identifier": [vnfd_id, vnf_instance_id],
 481             "operationStates": ["ANY"],
 482             "operationTypes": ["ANY"],
 483             "notificationType": notification_type,
 484         }
 485         if op_state:
 486             filter_q["operationStates"].append(op_state)
 487         if command:
 488             filter_q["operationTypes"].append(command)
 489         subscribers = []
 490         try:
 491             subscribers = self.db.get_list(self.subscriber_collection, filter_q)
 492             subscribers = self._format_vnflcm_subscribers(subscribers, event_details)
 493         except Exception as e:
 494             error_text = type(e).__name__ + ": " + str(e)
 495             self.logger.debug("Error getting vnflcm subscribers: {}".format(error_text))
 496         finally:
 497             return subscribers
 498
 499
 500 class NsdNotification(NotificationBase):
 501     def __init__(self, db):
 502         """
 503         Constructor of the class
 504         """
 505         super().__init__(db)
 506         # TODO will update this once support is there from subscription
 507         self.response_models = {}
 508         self.subscriber_collection = None
 509
 510
 511 class VnfdNotification(NotificationBase):
 512     def __init__(self, db):
 513         """
 514         Constructor of the class
 515         """
 516         super().__init__(db)
 517         # TODO will update this once support is there from subscription
 518         self.response_models = {}
 519         self.subscriber_collection = None