osm_nbi/notifications.py - osm/NBI - Gitiles

 # Copyright 2020 K Sai Kiran (Tata Elxsi)
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
 #
 #    http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
 # implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.

 __author__ = "K Sai Kiran <saikiran.k@tataelxsi.co.in>"
 __date__ = "$28-Apr-2020 23:59:59$"

 import asyncio
 import aiohttp
 from http import HTTPStatus
 import json
 import logging
 import time
 from uuid import uuid4


 class NotificationException(Exception):
     """
     Notification Exception
     """

     def __init__(self, message: str, http_code: int = HTTPStatus.BAD_REQUEST) -> None:
         """
         Constructor of notification exception
         :param message: String text containing exception details.
         :param http_code: HTTP status code of exception.
         """
         self.http_code = http_code
         Exception.__init__(self, message)


 class NotificationBase:
     response_models = None
     # Common HTTP payload header for all notifications.
     payload_header = {"Content-Type": "application/json", "Accept": "application/json"}

     def __init__(self, db) -> None:
         """
         Constructor of NotificationBase class.
         :param db: Database handler.
         """
         self.db = db
         self.logger = logging.getLogger("nbi.notifications")
         self.subscriber_collection = None

     def get_models(self) -> dict:
         """
         Returns the SOL005 model of notification class
         :param None
         :return: dict of SOL005 data model
         """
         return NotificationBase.response_models

     def get_subscribers(self, **kwargs) -> NotificationException:
         """
         Method should be implemented by all notification subclasses
         :param kwargs: any keyword arguments needed for db query.
         :return: List of subscribers
         """
         raise NotificationException(
             "Method get_subscribers() is not implemented",
             http_code=HTTPStatus.NOT_IMPLEMENTED,
         )

     @staticmethod
     def _get_basic_auth(username: str, password: str) -> tuple:
         return aiohttp.BasicAuth(username, password)

     def _decrypt_password(
         self, hashed: str, salt: str, schema_version: str = "1.1"
     ) -> str:
         return self.db.decrypt(hashed, schema_version, salt=salt)

     def get_payload(self, meta_notification: dict) -> dict:
         """
         Generates SOL005 compliant payload structure and returns them in dictionary.
         :param meta_notification: notification meta data which needs to be formatted as SOL005 compliant
         :return: A dictionary which is SOL005 compliant.
         """
         model_name = meta_notification["notificationType"]
         response_models = self.get_models()
         if not response_models or not response_models.get(model_name):
             raise NotificationException(
                 "Response model {} is not defined.".format(model_name),
                 HTTPStatus.NOT_IMPLEMENTED,
             )
         model_keys = response_models[model_name]
         payload = dict.fromkeys(model_keys, "N/A")
         notification_keys = set(meta_notification.keys())
         for model_key in model_keys.intersection(notification_keys):
             payload[model_key] = meta_notification[model_key]
         self.logger.debug(
             "Payload generated for subscriber: {} for {}".format(
                 payload["subscriptionId"], payload["notificationType"]
             )
         )
         return payload

     async def send_notifications(
         self,
         subscribers: list,
     ):
         """
         Generate tasks for all notification for an event.
         :param subscribers: A list of subscribers who want to be notified for event.
         """
         notifications = []
         for subscriber in subscribers:
             # Notify without auth
             if not subscriber.get("authentication"):
                 notifications.append(
                     {
                         "headers": self.payload_header,
                         "payload": self.get_payload(subscriber),
                         "CallbackUri": subscriber["CallbackUri"],
                     }
                 )
             elif subscriber["authentication"]["authType"] == "basic":
                 salt = subscriber["subscriptionId"]
                 hashed_password = subscriber["authentication"]["paramsBasic"][
                     "password"
                 ]
                 password = self._decrypt_password(hashed_password, salt)
                 auth_basic = self._get_basic_auth(
                     subscriber["authentication"]["paramsBasic"]["userName"], password
                 )
                 notifications.append(
                     {
                         "headers": self.payload_header,
                         "payload": self.get_payload(subscriber),
                         "auth_basic": auth_basic,
                         "CallbackUri": subscriber["CallbackUri"],
                     }
                 )
             # TODO add support for AuthType OAuth and TLS after support is added in subscription.
             else:
                 self.logger.debug(
                     "Subscriber {} can not be notified. {} notification auth type is not implemented".format(
                         subscriber["subscriptionId"],
                         subscriber["authentication"]["authType"],
                     )
                 )

         if notifications:
             tasks = []
             async with aiohttp.ClientSession() as session:
                 for notification in notifications:
                     tasks.append(
                         asyncio.ensure_future(
                             self.send_notification(session, notification),
                         )
                     )
                 await asyncio.gather(*tasks)

     async def send_notification(
         self,
         session: aiohttp.ClientSession,
         notification: dict,
         retry_count: int = 5,
         timeout: float = 5.0,
     ):
         """
         Performs HTTP Post request to notify subscriber. In case if for any reason notification is not sent successfully
         after maximum number of reties, then notification is dropped.
         :param session: An aiohttp client session object to maintain http session.
         :param notification: A dictionary containing all necessary data to make POST request.
         :param retry_count: An integer specifying the maximum number of reties for a notification.
         :param timeout: A float representing client timeout of each HTTP request.
         """
         backoff_delay = 1
         while retry_count > 0:
             try:
                 async with session.post(
                     url=notification["CallbackUri"],
                     headers=notification["headers"],
                     auth=notification.get("auth_basic", None),
                     data=json.dumps(notification["payload"]),
                     timeout=timeout,
                 ) as resp:
                     # self.logger.debug("Notification response: {}".format(resp.status))
                     if resp.status == HTTPStatus.NO_CONTENT:
                         self.logger.debug(
                             "Notification sent successfully to subscriber {}".format(
                                 notification["payload"]["subscriptionId"]
                             )
                         )
                     else:
                         error_text = "Erroneous response code: {}, ".format(resp.status)
                         error_text += await resp.text()
                         raise NotificationException(error_text)
                 return True
             except Exception as e:
                 error_text = type(e).__name__ + ": " + str(e)
                 self.logger.debug(
                     "Unable to send notification to subscriber {}. Details: {}".format(
                         notification["payload"]["subscriptionId"], error_text
                     )
                 )
                 error_detail = {
                     "error": type(e).__name__,
                     "error_text": str(e),
                     "timestamp": time.time(),
                 }
                 if "error_details" in notification["payload"].keys():
                     notification["payload"]["error_details"].append(error_detail)
                 else:
                     notification["payload"]["error_details"] = [error_detail]
                 retry_count -= 1
                 backoff_delay *= 2
                 self.logger.debug(
                     "Retry Notification for subscriber: {} after backoff delay: {} seconds.".format(
                         notification["payload"]["subscriptionId"], backoff_delay
                     )
                 )
                 await asyncio.sleep(backoff_delay)
         # Dropping notification
         self.logger.debug(
             "Notification {} sent failed to subscriber:{}.".format(
                 notification["payload"]["notificationType"],
                 notification["payload"]["subscriptionId"],
             )
         )
         return False


 class NsLcmNotification(NotificationBase):
     # maps kafka commands of completed operations to the original operation type
     completed_operation_map = {
         "INSTANTIATED": "INSTANTIATE",
         "SCALED": "SCALE",
         "TERMINATED": "TERMINATE",
         "UPDATED": "UPDATE",
         "HEALED": "HEAL",
     }
     # SOL005 response model for nslcm notifications
     response_models = {
         "NsLcmOperationOccurrenceNotification": {
             "id",
             "nsInstanceId",
             "nsLcmOpOccId",
             "operation",
             "notificationType",
             "subscriptionId",
             "timestamp",
             "notificationStatus",
             "operationState",
             "isAutomaticInvocation",
             "affectedVnf",
             "affectedVl",
             "affectedVnffg",
             "affectedNs",
             "affectedSap",
             "error",
             "_links",
         },
         "NsIdentifierCreationNotification": {
             "notificationType",
             "subscriptionId",
             "timestamp",
             "nsInstanceId",
             "_links",
         },
         "NsIdentifierDeletionNotification": {
             "notificationType",
             "subscriptionId",
             "timestamp",
             "nsInstanceId",
             "_links",
         },
         "NsChangeNotification": {
             "nsInstanceId",
             "nsComponentType",
             "nsComponentId",
             "lcmOpOccIdImpactngNsComponent",
             "lcmOpNameImpactingNsComponent",
             "lcmOpOccStatusImpactingNsComponent",
             "notificationType",
             "subscriptionId",
             "timeStamp",
             "error",
             "_links",
         },
     }

     def __init__(self, db) -> None:
         """
         Constructor of NsLcmNotification class.
         :param db: Database handler.
         """
         super().__init__(db)
         self.subscriber_collection = "mapped_subscriptions"

     def get_models(self) -> dict:
         """
         Returns the SOL005 model of notification class
         :param None
         :return: dict of SOL005 data model
         """
         return NsLcmNotification.response_models

     @staticmethod
     def _format_nslcm_subscribers(subscribers: list, event_details: dict) -> list:
         """
         Formats the raw event details from kakfa message and subscriber details.
         :param subscribers: A list of subscribers whom the event needs to be notified.
         :param event_details: A dict containing all meta data of event.
         :return:
         """
         notification_id = str(uuid4())
         event_timestamp = event_details["params"]["startTime"]
         resource_links = event_details["params"]["links"]
         event_operation = event_details["command"]
         for key in ["_admin", "_id", "id", "links"]:
             event_details["params"].pop(key, None)
         for subscriber in subscribers:
             subscriber["id"] = notification_id
             subscriber["timestamp"] = event_timestamp
             subscriber["_links"] = resource_links
             subscriber["subscriptionId"] = subscriber["reference"]
             subscriber["operation"] = event_operation
             del subscriber["reference"]
             del subscriber["_id"]
             subscriber.update(event_details["params"])
         return subscribers

     def get_subscribers(
         self,
         nsd_id: str,
         ns_instance_id: str,
         command: str,
         op_state: str,
         event_details: dict,
     ) -> list:
         """
         Queries database and returns list of subscribers.
         :param nsd_id: NSD id of an NS whose lifecycle has changed. (scaled, terminated. etc)
         :param ns_instance_id: NS instance id an NS whose lifecycle has changed.
         :param command: the command for event.
         :param op_state: the operation state of NS.
         :param event_details: dict containing raw data of event occured.
         :return: List of interested subscribers for occurred event.
         """
         notification_type = [
             "NsLcmOperationOccurrenceNotification",
             "NsChangeNotification",
             "NsIdentifierCreationNotification",
             "NsIdentifierDeletionNotification",
         ]
         filter_q = {
             "identifier": [nsd_id, ns_instance_id],
             "operationStates": ["ANY"],
             "operationTypes": ["ANY"],
             "notificationType": notification_type,
         }
         if op_state:
             filter_q["operationStates"].append(op_state)
         if command:
             op_type = self.completed_operation_map.get(command, command)
             filter_q["operationTypes"].append(op_type)
         # self.logger.debug("Db query is: {}".format(filter_q))
         subscribers = []
         try:
             subscribers = self.db.get_list(self.subscriber_collection, filter_q)
             subscribers = self._format_nslcm_subscribers(subscribers, event_details)
         except Exception as e:
             error_text = type(e).__name__ + ": " + str(e)
             self.logger.debug("Error getting nslcm subscribers: {}".format(error_text))
         finally:
             return subscribers


 class VnfLcmNotification(NotificationBase):
     # SOL003 response model for vnflcm notifications
     response_models = {
         "VnfLcmOperationOccurrenceNotification": {
             "id",
             "notificationType",
             "subscriptionId",
             "timeStamp",
             "notificationStatus",
             "operationState",
             "vnfInstanceId",
             "operation",
             "isAutomaticInvocation",
             "vnfLcmOpOccId",
             "affectedVnfcs",
             "affectedVirtualLinks",
             "affectedExtLinkPorts",
             "affectedVirtualStorages",
             "changedInfo",
             "changedExtConnectivity",
             "modificationsTriggeredByVnfPkgChange",
             "error",
             "_links",
         },
         "VnfIdentifierCreationNotification": {
             "id",
             "notificationType",
             "subscriptionId",
             "timeStamp",
             "vnfInstanceId",
             "_links",
         },
         "VnfIdentifierDeletionNotification": {
             "id",
             "notificationType",
             "subscriptionId",
             "timeStamp",
             "vnfInstanceId",
             "_links",
         },
     }

     def __init__(self, db) -> None:
         """
         Constructor of VnfLcmNotification class.
         :param db: Database handler.
         """
         super().__init__(db)
         self.subscriber_collection = "mapped_subscriptions"

     def get_models(self) -> dict:
         """
         Returns the SOL003 model of notification class
         :param None
         :return: dict of SOL003 data model
         """
         return self.response_models

     def _format_vnflcm_subscribers(
         self, subscribers: list, event_details: dict
     ) -> list:
         """
         Formats the raw event details from kafka message and subscriber details.
         :param subscribers: A list of subscribers whom the event needs to be notified.
         :param event_details: A dict containing all meta data of event.
         :return:
         """
         notification_id = str(uuid4())
         event_timestamp = time.time()
         event_operation = event_details["command"]
         for subscriber in subscribers:
             subscriber["id"] = notification_id
             subscriber["timeStamp"] = event_timestamp
             subscriber["subscriptionId"] = subscriber["reference"]
             subscriber["operation"] = event_operation
             del subscriber["reference"]
             del subscriber["_id"]
             subscriber.update(event_details["params"])
         return subscribers

     def get_subscribers(
         self,
         vnfd_id: str,
         vnf_instance_id: str,
         command: str,
         op_state: str,
         event_details: dict,
     ) -> list:
         """
         Queries database and returns list of subscribers.
         :param vnfd_id: Vnfd id of a VNF whose lifecycle has changed. (instantiated, scaled, terminated. etc)
         :param vnf_instance_id: Vnf instance id of a VNF whose lifecycle has changed.
         :param command: the command for event.
         :param op_state: the operation state of VNF.
         :param event_details: dict containing raw data of event occurred.
         :return: List of interested subscribers for occurred event.
         """
         notification_type = [
             "VnfIdentifierCreationNotification",
             "VnfLcmOperationOccurrenceNotification",
             "VnfIdentifierDeletionNotification",
         ]
         filter_q = {
             "identifier": [vnfd_id, vnf_instance_id],
             "operationStates": ["ANY"],
             "operationTypes": ["ANY"],
             "notificationType": notification_type,
         }
         if op_state:
             filter_q["operationStates"].append(op_state)
         if command:
             filter_q["operationTypes"].append(command)
         subscribers = []
         try:
             subscribers = self.db.get_list(self.subscriber_collection, filter_q)
             subscribers = self._format_vnflcm_subscribers(subscribers, event_details)
         except Exception as e:
             error_text = type(e).__name__ + ": " + str(e)
             self.logger.debug("Error getting vnflcm subscribers: {}".format(error_text))
         finally:
             return subscribers


 class NsdNotification(NotificationBase):
     def __init__(self, db):
         """
         Constructor of the class
         """
         super().__init__(db)
         # TODO will update this once support is there from subscription
         self.response_models = {}
         self.subscriber_collection = None


 class VnfdNotification(NotificationBase):
     def __init__(self, db):
         """
         Constructor of the class
         """
         super().__init__(db)
         # TODO will update this once support is there from subscription
         self.response_models = {}
         self.subscriber_collection = None
	# Copyright 2020 K Sai Kiran (Tata Elxsi)
	#
	# Licensed under the Apache License, Version 2.0 (the "License");
	# you may not use this file except in compliance with the License.
	# You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing, software
	# distributed under the License is distributed on an "AS IS" BASIS,
	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
	# implied.
	# See the License for the specific language governing permissions and
	# limitations under the License.

	__author__ = "K Sai Kiran <saikiran.k@tataelxsi.co.in>"
	__date__ = "$28-Apr-2020 23:59:59$"

	import asyncio
	import aiohttp
	from http import HTTPStatus
	import json
	import logging
	import time
	from uuid import uuid4


	class NotificationException(Exception):
	"""
	Notification Exception
	"""

	def __init__(self, message: str, http_code: int = HTTPStatus.BAD_REQUEST) -> None:
	"""
	Constructor of notification exception
	:param message: String text containing exception details.
	:param http_code: HTTP status code of exception.
	"""
	self.http_code = http_code
	Exception.__init__(self, message)


	class NotificationBase:
	response_models = None
	# Common HTTP payload header for all notifications.
	payload_header = {"Content-Type": "application/json", "Accept": "application/json"}

	def __init__(self, db) -> None:
	"""
	Constructor of NotificationBase class.
	:param db: Database handler.
	"""
	self.db = db
	self.logger = logging.getLogger("nbi.notifications")
	self.subscriber_collection = None

	def get_models(self) -> dict:
	"""
	Returns the SOL005 model of notification class
	:param None
	:return: dict of SOL005 data model
	"""
	return NotificationBase.response_models

	def get_subscribers(self, **kwargs) -> NotificationException:
	"""
	Method should be implemented by all notification subclasses
	:param kwargs: any keyword arguments needed for db query.
	:return: List of subscribers
	"""
	raise NotificationException(
	"Method get_subscribers() is not implemented",
	http_code=HTTPStatus.NOT_IMPLEMENTED,
	)

	@staticmethod
	def _get_basic_auth(username: str, password: str) -> tuple:
	return aiohttp.BasicAuth(username, password)

	def _decrypt_password(
	self, hashed: str, salt: str, schema_version: str = "1.1"
	) -> str:
	return self.db.decrypt(hashed, schema_version, salt=salt)

	def get_payload(self, meta_notification: dict) -> dict:
	"""
	Generates SOL005 compliant payload structure and returns them in dictionary.
	:param meta_notification: notification meta data which needs to be formatted as SOL005 compliant
	:return: A dictionary which is SOL005 compliant.
	"""
	model_name = meta_notification["notificationType"]
	response_models = self.get_models()
	if not response_models or not response_models.get(model_name):
	raise NotificationException(
	"Response model {} is not defined.".format(model_name),
	HTTPStatus.NOT_IMPLEMENTED,
	)
	model_keys = response_models[model_name]
	payload = dict.fromkeys(model_keys, "N/A")
	notification_keys = set(meta_notification.keys())
	for model_key in model_keys.intersection(notification_keys):
	payload[model_key] = meta_notification[model_key]
	self.logger.debug(
	"Payload generated for subscriber: {} for {}".format(
	payload["subscriptionId"], payload["notificationType"]
	)
	)
	return payload

	async def send_notifications(
	self,
	subscribers: list,
	):
	"""
	Generate tasks for all notification for an event.
	:param subscribers: A list of subscribers who want to be notified for event.
	"""
	notifications = []
	for subscriber in subscribers:
	# Notify without auth
	if not subscriber.get("authentication"):
	notifications.append(
	{
	"headers": self.payload_header,
	"payload": self.get_payload(subscriber),
	"CallbackUri": subscriber["CallbackUri"],
	}
	)
	elif subscriber["authentication"]["authType"] == "basic":
	salt = subscriber["subscriptionId"]
	hashed_password = subscriber["authentication"]["paramsBasic"][
	"password"
	]
	password = self._decrypt_password(hashed_password, salt)
	auth_basic = self._get_basic_auth(
	subscriber["authentication"]["paramsBasic"]["userName"], password
	)
	notifications.append(
	{
	"headers": self.payload_header,
	"payload": self.get_payload(subscriber),
	"auth_basic": auth_basic,
	"CallbackUri": subscriber["CallbackUri"],
	}
	)
	# TODO add support for AuthType OAuth and TLS after support is added in subscription.
	else:
	self.logger.debug(
	"Subscriber {} can not be notified. {} notification auth type is not implemented".format(
	subscriber["subscriptionId"],
	subscriber["authentication"]["authType"],
	)
	)

	if notifications:
	tasks = []
	async with aiohttp.ClientSession() as session:
	for notification in notifications:
	tasks.append(
	asyncio.ensure_future(
	self.send_notification(session, notification),
	)
	)
	await asyncio.gather(*tasks)

	async def send_notification(
	self,
	session: aiohttp.ClientSession,
	notification: dict,
	retry_count: int = 5,
	timeout: float = 5.0,
	):
	"""
	Performs HTTP Post request to notify subscriber. In case if for any reason notification is not sent successfully
	after maximum number of reties, then notification is dropped.
	:param session: An aiohttp client session object to maintain http session.
	:param notification: A dictionary containing all necessary data to make POST request.
	:param retry_count: An integer specifying the maximum number of reties for a notification.
	:param timeout: A float representing client timeout of each HTTP request.
	"""
	backoff_delay = 1
	while retry_count > 0:
	try:
	async with session.post(
	url=notification["CallbackUri"],
	headers=notification["headers"],
	auth=notification.get("auth_basic", None),
	data=json.dumps(notification["payload"]),
	timeout=timeout,
	) as resp:
	# self.logger.debug("Notification response: {}".format(resp.status))
	if resp.status == HTTPStatus.NO_CONTENT:
	self.logger.debug(
	"Notification sent successfully to subscriber {}".format(
	notification["payload"]["subscriptionId"]
	)
	)
	else:
	error_text = "Erroneous response code: {}, ".format(resp.status)
	error_text += await resp.text()
	raise NotificationException(error_text)
	return True
	except Exception as e:
	error_text = type(e).__name__ + ": " + str(e)
	self.logger.debug(
	"Unable to send notification to subscriber {}. Details: {}".format(
	notification["payload"]["subscriptionId"], error_text
	)
	)
	error_detail = {
	"error": type(e).__name__,
	"error_text": str(e),
	"timestamp": time.time(),
	}
	if "error_details" in notification["payload"].keys():
	notification["payload"]["error_details"].append(error_detail)
	else:
	notification["payload"]["error_details"] = [error_detail]
	retry_count -= 1
	backoff_delay *= 2
	self.logger.debug(
	"Retry Notification for subscriber: {} after backoff delay: {} seconds.".format(
	notification["payload"]["subscriptionId"], backoff_delay
	)
	)
	await asyncio.sleep(backoff_delay)
	# Dropping notification
	self.logger.debug(
	"Notification {} sent failed to subscriber:{}.".format(
	notification["payload"]["notificationType"],
	notification["payload"]["subscriptionId"],
	)
	)
	return False


	class NsLcmNotification(NotificationBase):
	# maps kafka commands of completed operations to the original operation type
	completed_operation_map = {
	"INSTANTIATED": "INSTANTIATE",
	"SCALED": "SCALE",
	"TERMINATED": "TERMINATE",
	"UPDATED": "UPDATE",
	"HEALED": "HEAL",
	}
	# SOL005 response model for nslcm notifications
	response_models = {
	"NsLcmOperationOccurrenceNotification": {
	"id",
	"nsInstanceId",
	"nsLcmOpOccId",
	"operation",
	"notificationType",
	"subscriptionId",
	"timestamp",
	"notificationStatus",
	"operationState",
	"isAutomaticInvocation",
	"affectedVnf",
	"affectedVl",
	"affectedVnffg",
	"affectedNs",
	"affectedSap",
	"error",
	"_links",
	},
	"NsIdentifierCreationNotification": {
	"notificationType",
	"subscriptionId",
	"timestamp",
	"nsInstanceId",
	"_links",
	},
	"NsIdentifierDeletionNotification": {
	"notificationType",
	"subscriptionId",
	"timestamp",
	"nsInstanceId",
	"_links",
	},
	"NsChangeNotification": {
	"nsInstanceId",
	"nsComponentType",
	"nsComponentId",
	"lcmOpOccIdImpactngNsComponent",
	"lcmOpNameImpactingNsComponent",
	"lcmOpOccStatusImpactingNsComponent",
	"notificationType",
	"subscriptionId",
	"timeStamp",
	"error",
	"_links",
	},
	}

	def __init__(self, db) -> None:
	"""
	Constructor of NsLcmNotification class.
	:param db: Database handler.
	"""
	super().__init__(db)
	self.subscriber_collection = "mapped_subscriptions"

	def get_models(self) -> dict:
	"""
	Returns the SOL005 model of notification class
	:param None
	:return: dict of SOL005 data model
	"""
	return NsLcmNotification.response_models

	@staticmethod
	def _format_nslcm_subscribers(subscribers: list, event_details: dict) -> list:
	"""
	Formats the raw event details from kakfa message and subscriber details.
	:param subscribers: A list of subscribers whom the event needs to be notified.
	:param event_details: A dict containing all meta data of event.
	:return:
	"""
	notification_id = str(uuid4())
	event_timestamp = event_details["params"]["startTime"]
	resource_links = event_details["params"]["links"]
	event_operation = event_details["command"]
	for key in ["_admin", "_id", "id", "links"]:
	event_details["params"].pop(key, None)
	for subscriber in subscribers:
	subscriber["id"] = notification_id
	subscriber["timestamp"] = event_timestamp
	subscriber["_links"] = resource_links
	subscriber["subscriptionId"] = subscriber["reference"]
	subscriber["operation"] = event_operation
	del subscriber["reference"]
	del subscriber["_id"]
	subscriber.update(event_details["params"])
	return subscribers

	def get_subscribers(
	self,
	nsd_id: str,
	ns_instance_id: str,
	command: str,
	op_state: str,
	event_details: dict,
	) -> list:
	"""
	Queries database and returns list of subscribers.
	:param nsd_id: NSD id of an NS whose lifecycle has changed. (scaled, terminated. etc)
	:param ns_instance_id: NS instance id an NS whose lifecycle has changed.
	:param command: the command for event.
	:param op_state: the operation state of NS.
	:param event_details: dict containing raw data of event occured.
	:return: List of interested subscribers for occurred event.
	"""
	notification_type = [
	"NsLcmOperationOccurrenceNotification",
	"NsChangeNotification",
	"NsIdentifierCreationNotification",
	"NsIdentifierDeletionNotification",
	]
	filter_q = {
	"identifier": [nsd_id, ns_instance_id],
	"operationStates": ["ANY"],
	"operationTypes": ["ANY"],
	"notificationType": notification_type,
	}
	if op_state:
	filter_q["operationStates"].append(op_state)
	if command:
	op_type = self.completed_operation_map.get(command, command)
	filter_q["operationTypes"].append(op_type)
	# self.logger.debug("Db query is: {}".format(filter_q))
	subscribers = []
	try:
	subscribers = self.db.get_list(self.subscriber_collection, filter_q)
	subscribers = self._format_nslcm_subscribers(subscribers, event_details)
	except Exception as e:
	error_text = type(e).__name__ + ": " + str(e)
	self.logger.debug("Error getting nslcm subscribers: {}".format(error_text))
	finally:
	return subscribers


	class VnfLcmNotification(NotificationBase):
	# SOL003 response model for vnflcm notifications
	response_models = {
	"VnfLcmOperationOccurrenceNotification": {
	"id",
	"notificationType",
	"subscriptionId",
	"timeStamp",
	"notificationStatus",
	"operationState",
	"vnfInstanceId",
	"operation",
	"isAutomaticInvocation",
	"vnfLcmOpOccId",
	"affectedVnfcs",
	"affectedVirtualLinks",
	"affectedExtLinkPorts",
	"affectedVirtualStorages",
	"changedInfo",
	"changedExtConnectivity",
	"modificationsTriggeredByVnfPkgChange",
	"error",
	"_links",
	},
	"VnfIdentifierCreationNotification": {
	"id",
	"notificationType",
	"subscriptionId",
	"timeStamp",
	"vnfInstanceId",
	"_links",
	},
	"VnfIdentifierDeletionNotification": {
	"id",
	"notificationType",
	"subscriptionId",
	"timeStamp",
	"vnfInstanceId",
	"_links",
	},
	}

	def __init__(self, db) -> None:
	"""
	Constructor of VnfLcmNotification class.
	:param db: Database handler.
	"""
	super().__init__(db)
	self.subscriber_collection = "mapped_subscriptions"

	def get_models(self) -> dict:
	"""
	Returns the SOL003 model of notification class
	:param None
	:return: dict of SOL003 data model
	"""
	return self.response_models

	def _format_vnflcm_subscribers(
	self, subscribers: list, event_details: dict
	) -> list:
	"""
	Formats the raw event details from kafka message and subscriber details.
	:param subscribers: A list of subscribers whom the event needs to be notified.
	:param event_details: A dict containing all meta data of event.
	:return:
	"""
	notification_id = str(uuid4())
	event_timestamp = time.time()
	event_operation = event_details["command"]
	for subscriber in subscribers:
	subscriber["id"] = notification_id
	subscriber["timeStamp"] = event_timestamp
	subscriber["subscriptionId"] = subscriber["reference"]
	subscriber["operation"] = event_operation
	del subscriber["reference"]
	del subscriber["_id"]
	subscriber.update(event_details["params"])
	return subscribers

	def get_subscribers(
	self,
	vnfd_id: str,
	vnf_instance_id: str,
	command: str,
	op_state: str,
	event_details: dict,
	) -> list:
	"""
	Queries database and returns list of subscribers.
	:param vnfd_id: Vnfd id of a VNF whose lifecycle has changed. (instantiated, scaled, terminated. etc)
	:param vnf_instance_id: Vnf instance id of a VNF whose lifecycle has changed.
	:param command: the command for event.
	:param op_state: the operation state of VNF.
	:param event_details: dict containing raw data of event occurred.
	:return: List of interested subscribers for occurred event.
	"""
	notification_type = [
	"VnfIdentifierCreationNotification",
	"VnfLcmOperationOccurrenceNotification",
	"VnfIdentifierDeletionNotification",
	]
	filter_q = {
	"identifier": [vnfd_id, vnf_instance_id],
	"operationStates": ["ANY"],
	"operationTypes": ["ANY"],
	"notificationType": notification_type,
	}
	if op_state:
	filter_q["operationStates"].append(op_state)
	if command:
	filter_q["operationTypes"].append(command)
	subscribers = []
	try:
	subscribers = self.db.get_list(self.subscriber_collection, filter_q)
	subscribers = self._format_vnflcm_subscribers(subscribers, event_details)
	except Exception as e:
	error_text = type(e).__name__ + ": " + str(e)
	self.logger.debug("Error getting vnflcm subscribers: {}".format(error_text))
	finally:
	return subscribers


	class NsdNotification(NotificationBase):
	def __init__(self, db):
	"""
	Constructor of the class
	"""
	super().__init__(db)
	# TODO will update this once support is there from subscription
	self.response_models = {}
	self.subscriber_collection = None


	class VnfdNotification(NotificationBase):
	def __init__(self, db):
	"""
	Constructor of the class
	"""
	super().__init__(db)
	# TODO will update this once support is there from subscription
	self.response_models = {}
	self.subscriber_collection = None