X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=o2ims%2Fservice%2Fcommand%2Fnotify_handler.py;h=08da45ded9fc44bff865f28c1634f45e77175e12;hb=8d6c2f135ee103b9a94ea6c29986a608069b6982;hp=8f134aff32944bc2ce9352b1094139b70d26114a;hpb=45f6aa19191f366a05aa6df0902536c225c6d970;p=pti%2Fo2.git diff --git a/o2ims/service/command/notify_handler.py b/o2ims/service/command/notify_handler.py index 8f134af..08da45d 100644 --- a/o2ims/service/command/notify_handler.py +++ b/o2ims/service/command/notify_handler.py @@ -12,20 +12,24 @@ # See the License for the specific language governing permissions and # limitations under the License. -import json # import redis # import requests +import json +import ssl from urllib.parse import urlparse # from o2common.config import config +from o2common.domain.filter import gen_orm_filter from o2common.service.unit_of_work import AbstractUnitOfWork -from o2ims.domain import commands -from o2ims.domain.subscription_obj import Subscription, Message2SMO from o2common.service.command.handler import get_https_conn_default from o2common.service.command.handler import get_http_conn from o2common.service.command.handler import get_https_conn_selfsigned from o2common.service.command.handler import post_data -import ssl + +from o2ims.domain import commands, ocloud +from o2ims.domain.subscription_obj import Subscription, Message2SMO, \ + NotificationEventEnum + from o2common.helper import o2logging logger = o2logging.get_logger(__name__) @@ -38,57 +42,211 @@ def notify_change_to_smo( cmd: commands.PubMessage2SMO, uow: AbstractUnitOfWork, ): - logger.info('In notify_change_to_smo') + logger.debug('In notify_change_to_smo') + msg_type = cmd.type + if msg_type == 'ResourceType': + _notify_resourcetype(uow, cmd) + elif msg_type == 'ResourcePool': + _notify_resourcepool(uow, cmd) + elif msg_type == 'Dms': + _notify_dms(uow, cmd) + elif msg_type == 'Resource': + _notify_resource(uow, cmd) + + +def _notify_resourcetype(uow, cmd): data = cmd.data + msg_type = cmd.type with uow: + resource_type = uow.resource_types.get(data.id) + if resource_type is None: + logger.warning('ResourceType {} does not exists.'.format(data.id)) + return + resource_type_dict = { + 'resourceTypeId': resource_type.resourceTypeId, + 'name': resource_type.name, + 'description': resource_type.description, + 'vendor': resource_type.vendor, + 'model': resource_type.model, + 'version': resource_type.version, + # 'alarmDictionary': resource_type.alarmDictionary.serialize() + } + subs = uow.subscriptions.list() for sub in subs: sub_data = sub.serialize() logger.debug('Subscription: {}'.format(sub_data['subscriptionId'])) + if not sub_data.get('filter', None): + callback_smo(sub, data, msg_type, resource_type_dict) + continue + try: + args = gen_orm_filter(ocloud.ResourceType, sub_data['filter']) + except KeyError: + logger.warning( + 'Subscription {} filter {} has wrong attribute name ' + 'or value. Ignore the filter.'.format( + sub_data['subscriptionId'], sub_data['filter'])) + callback_smo(sub, data, msg_type, resource_type_dict) + continue + args.append(ocloud.ResourceType.resourceTypeId == data.id) + ret = uow.resource_types.list_with_count(*args) + if ret[0] != 0: + logger.debug( + 'ResourceType {} skip for subscription {} because of the ' + 'filter.' + .format(data.id, sub_data['subscriptionId'])) + continue + callback_smo(sub, data, msg_type, resource_type_dict) + +def _notify_resourcepool(uow, cmd): + data = cmd.data + msg_type = cmd.type + with uow: + resource_pool = uow.resource_pools.get(data.id) + if resource_pool is None: + logger.warning('ResourcePool {} does not exists.'.format(data.id)) + return + resource_pool_dict = { + 'resourcePoolId': resource_pool.resourcePoolId, + 'oCloudId': resource_pool.oCloudId, + 'globalLocationId': resource_pool.globalLocationId, + 'name': resource_pool.name, + 'description': resource_pool.description + } + + subs = uow.subscriptions.list() + for sub in subs: + sub_data = sub.serialize() + logger.debug('Subscription: {}'.format(sub_data['subscriptionId'])) + if not sub_data.get('filter', None): + callback_smo(sub, data, msg_type, resource_pool_dict) + continue try: - resource_filter = json.loads(sub_data['filter']) - if len(resource_filter) > 0: - resource = uow.resources.get(data.id) - logger.debug(type(resource)) - if resource: # TODO deal with resource is empty - res_type_id = resource.serialize()['resourceTypeId'] - resourcetype = uow.resource_types.get(res_type_id) - logger.debug(resourcetype.name) - if resourcetype.name not in resource_filter: - continue - except json.decoder.JSONDecodeError as err: + args = gen_orm_filter(ocloud.Resource, sub_data['filter']) + except KeyError: logger.warning( - 'subscription filter decode json failed: {}'.format(err)) + 'Subscription {} filter {} has wrong attribute name ' + 'or value. Ignore the filter.'.format( + sub_data['subscriptionId'], sub_data['filter'])) + callback_smo(sub, data, msg_type, resource_pool_dict) + continue + args.append(ocloud.ResourcePool.resourcePoolId == data.id) + ret = uow.resource_pools.list_with_count(*args) + if ret[0] != 0: + logger.debug( + 'ResourcePool {} skip for subscription {} because of the ' + 'filter.' + .format(data.id, sub_data['subscriptionId'])) + continue + callback_smo(sub, data, msg_type, resource_pool_dict) + +def _notify_dms(uow, cmd): + data = cmd.data + msg_type = cmd.type + with uow: + dms = uow.deployment_managers.get(data.id) + if dms is None: + logger.warning( + 'DeploymentManager {} does not exists.'.format(data.id)) + return + dms_dict = { + 'deploymentManagerId': dms.deploymentManagerId, + 'name': dms.name, + 'description': dms.description, + 'oCloudId': dms.oCloudId, + 'serviceUri': dms.serviceUri + } + + subs = uow.subscriptions.list() + for sub in subs: + sub_data = sub.serialize() + logger.debug('Subscription: {}'.format(sub_data['subscriptionId'])) + if not sub_data.get('filter', None): + callback_smo(sub, data, msg_type, dms_dict) + continue + try: + args = gen_orm_filter(ocloud.Resource, sub_data['filter']) + except KeyError: + logger.warning( + 'Subscription {} filter {} has wrong attribute name ' + 'or value. Ignore the filter.'.format( + sub_data['subscriptionId'], sub_data['filter'])) + callback_smo(sub, data, msg_type, dms_dict) + continue + args.append( + ocloud.DeploymentManager.deploymentManagerId == data.id) + ret = uow.deployment_managers.list_with_count(*args) + if ret[0] != 0: + logger.debug( + 'DeploymentManager {} skip for subscription {} because of ' + 'the filter.' + .format(data.id, sub_data['subscriptionId'])) + continue callback_smo(sub, data) + callback_smo(sub, data, msg_type, dms_dict) + + +def _notify_resource(uow, cmd): + data = cmd.data + msg_type = cmd.type + with uow: + resource = uow.resources.get(data.id) + if resource is None: + logger.warning('Resource {} does not exists.'.format(data.id)) + return + res_pool_id = resource.serialize()['resourcePoolId'] + logger.debug('res pool id is {}'.format(res_pool_id)) + subs = uow.subscriptions.list() + for sub in subs: + sub_data = sub.serialize() + logger.debug('Subscription: {}'.format(sub_data['subscriptionId'])) + if not sub_data.get('filter', None): + callback_smo(sub, data, msg_type) + continue + try: + args = gen_orm_filter(ocloud.Resource, sub_data['filter']) + except KeyError: + logger.warning( + 'Subscription {} filter {} has wrong attribute name ' + 'or value. Ignore the filter.'.format( + sub_data['subscriptionId'], sub_data['filter'])) + callback_smo(sub, data, msg_type) + continue + args.append(ocloud.Resource.resourceId == data.id) + ret = uow.resources.list_with_count(res_pool_id, *args) + if ret[0] != 0: + logger.debug( + 'Resource {} skip for subscription {} because of the ' + 'filter.' + .format(data.id, sub_data['subscriptionId'])) + continue + callback_smo(sub, data, msg_type) -def callback_smo(sub: Subscription, msg: Message2SMO): + +def callback_smo(sub: Subscription, msg: Message2SMO, msg_type: str, + obj_dict: dict = None): sub_data = sub.serialize() - callback_data = json.dumps({ + callback = { 'consumerSubscriptionId': sub_data['consumerSubscriptionId'], 'notificationEventType': msg.notificationEventType, 'objectRef': msg.objectRef, 'updateTime': msg.updatetime - }) + } + if msg_type != 'Resource': + if msg.notificationEventType in [NotificationEventEnum.DELETE, + NotificationEventEnum.MODIFY]: + callback['priorObjectState'] = obj_dict + if msg.notificationEventType in [NotificationEventEnum.CREATE, + NotificationEventEnum.MODIFY]: + callback['postObjectState'] = obj_dict + callback_data = json.dumps(callback) logger.info('URL: {}, data: {}'.format( sub_data['callback'], callback_data)) - # r.publish(sub_data['subscriptionId'], json.dumps({ - # 'consumerSubscriptionId': sub_data['consumerSubscriptionId'], - # 'notificationEventType': msg.notificationEventType, - # 'objectRef': msg.objectRef - # })) - # try: - # headers = {'User-Agent': 'Mozilla/5.0'} - # resp = requests.post(sub_data['callback'], data=callback_data, - # headers=headers) - # if resp.status_code == 202 or resp.status_code == 200: - # logger.info('Notify to SMO successed') - # return - # logger.error('Response code is: {}'.format(resp.status_code)) - # except requests.exceptions.HTTPError as err: - # logger.error('request smo error: {}'.format(err)) + + # Call SMO through the SMO callback url o = urlparse(sub_data['callback']) if o.scheme == 'https': conn = get_https_conn_default(o.netloc)