o2ims/service/command/notify_handler.py

   1 # Copyright (C) 2021-2022 Wind River Systems, Inc.
   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 implied.
  12 #  See the License for the specific language governing permissions and
  13 #  limitations under the License.
  14
  15 # import redis
  16 # import requests
  17 import json
  18 import ssl
  19 from urllib.parse import urlparse
  20
  21 # from o2common.config import config
  22 from o2common.domain.filter import gen_orm_filter
  23 from o2common.service.unit_of_work import AbstractUnitOfWork
  24 from o2common.service.command.handler import get_https_conn_default
  25 from o2common.service.command.handler import get_http_conn
  26 from o2common.service.command.handler import get_https_conn_selfsigned
  27 from o2common.service.command.handler import post_data
  28
  29 from o2ims.domain import commands, ocloud
  30 from o2ims.domain.subscription_obj import Subscription, Message2SMO, \
  31     NotificationEventEnum
  32
  33 from o2common.helper import o2logging
  34 logger = o2logging.get_logger(__name__)
  35
  36
  37 # # Maybe another MQ server
  38 # r = redis.Redis(**config.get_redis_host_and_port())
  39
  40
  41 def notify_change_to_smo(
  42     cmd: commands.PubMessage2SMO,
  43     uow: AbstractUnitOfWork,
  44 ):
  45     logger.debug('In notify_change_to_smo')
  46     msg_type = cmd.type
  47     if msg_type == 'ResourceType':
  48         _notify_resourcetype(uow, cmd.data)
  49     elif msg_type == 'ResourcePool':
  50         _notify_resourcepool(uow, cmd.data)
  51     elif msg_type == 'Dms':
  52         _notify_dms(uow, cmd.data)
  53     elif msg_type == 'Resource':
  54         _notify_resource(uow, cmd.data)
  55
  56
  57 def _notify_resourcetype(uow, data):
  58     with uow:
  59         resource_type = uow.resource_types.get(data.id)
  60         if resource_type is None:
  61             logger.warning('ResourceType {} does not exists.'.format(data.id))
  62             return
  63         resource_type_dict = {
  64             'resourceTypeId': resource_type.resourceTypeId,
  65             'name': resource_type.name,
  66             'description': resource_type.description,
  67             'vendor': resource_type.vendor,
  68             'model': resource_type.model,
  69             'version': resource_type.version,
  70             # 'alarmDictionary': resource_type.alarmDictionary.serialize()
  71         }
  72
  73         subs = uow.subscriptions.list()
  74         for sub in subs:
  75             sub_data = sub.serialize()
  76             logger.debug('Subscription: {}'.format(sub_data['subscriptionId']))
  77             filters = handle_filter(sub_data['filter'], 'ResourceTypeInfo')
  78             if not filters:
  79                 callback_smo(sub, data, resource_type_dict)
  80                 continue
  81             filter_effect = 0
  82             for filter in filters:
  83                 try:
  84                     args = gen_orm_filter(ocloud.ResourceType, filter)
  85                 except KeyError:
  86                     logger.warning(
  87                         'Subscription {} filter {} has wrong attribute '
  88                         'name or value. Ignore the filter.'.format(
  89                             sub_data['subscriptionId'],
  90                             sub_data['filter']))
  91                     continue
  92                 if len(args) == 0 and 'objectType' in filter:
  93                     filter_effect += 1
  94                     break
  95                 args.append(ocloud.ResourceType.resourceTypeId == data.id)
  96                 ret = uow.resource_types.list_with_count(*args)
  97                 if ret[0] > 0:
  98                     logger.debug(
  99                         'ResourcePool {} skip for subscription {} because of'
 100                         ' the filter.'
 101                         .format(data.id, sub_data['subscriptionId']))
 102                     filter_effect += 1
 103                     break
 104             if filter_effect > 0:
 105                 continue
 106             callback_smo(sub, data, resource_type_dict)
 107
 108
 109 def _notify_resourcepool(uow, data):
 110     with uow:
 111         resource_pool = uow.resource_pools.get(data.id)
 112         if resource_pool is None:
 113             logger.warning('ResourcePool {} does not exists.'.format(data.id))
 114             return
 115         resource_pool_dict = {
 116             'resourcePoolId': resource_pool.resourcePoolId,
 117             'oCloudId': resource_pool.oCloudId,
 118             'globalLocationId': resource_pool.globalLocationId,
 119             'name': resource_pool.name,
 120             'description': resource_pool.description
 121         }
 122
 123         subs = uow.subscriptions.list()
 124         for sub in subs:
 125             sub_data = sub.serialize()
 126             logger.debug('Subscription: {}'.format(sub_data['subscriptionId']))
 127             filters = handle_filter(sub_data['filter'], 'ResourcePoolInfo')
 128             if not filters:
 129                 callback_smo(sub, data, resource_pool_dict)
 130                 continue
 131             filter_effect = 0
 132             for filter in filters:
 133                 try:
 134                     args = gen_orm_filter(ocloud.ResourcePool, filter)
 135                 except KeyError:
 136                     logger.warning(
 137                         'Subscription {} filter {} has wrong attribute '
 138                         'name or value. Ignore the filter.'.format(
 139                             sub_data['subscriptionId'],
 140                             sub_data['filter']))
 141                     continue
 142                 if len(args) == 0 and 'objectType' in filter:
 143                     filter_effect += 1
 144                     break
 145                 args.append(ocloud.ResourcePool.resourcePoolId == data.id)
 146                 ret = uow.resource_pools.list_with_count(*args)
 147                 if ret[0] > 0:
 148                     logger.debug(
 149                         'ResourcePool {} skip for subscription {} because of'
 150                         ' the filter.'
 151                         .format(data.id, sub_data['subscriptionId']))
 152                     filter_effect += 1
 153                     break
 154             if filter_effect > 0:
 155                 continue
 156             callback_smo(sub, data, resource_pool_dict)
 157
 158
 159 def _notify_dms(uow, data):
 160     with uow:
 161         dms = uow.deployment_managers.get(data.id)
 162         if dms is None:
 163             logger.warning(
 164                 'DeploymentManager {} does not exists.'.format(data.id))
 165             return
 166         dms_dict = {
 167             'deploymentManagerId': dms.deploymentManagerId,
 168             'name': dms.name,
 169             'description': dms.description,
 170             'oCloudId': dms.oCloudId,
 171             'serviceUri': dms.serviceUri
 172         }
 173
 174         subs = uow.subscriptions.list()
 175         for sub in subs:
 176             sub_data = sub.serialize()
 177             logger.debug('Subscription: {}'.format(sub_data['subscriptionId']))
 178             filters = handle_filter(
 179                 sub_data['filter'], 'DeploymentManagerInfo')
 180             if not filters:
 181                 callback_smo(sub, data, dms_dict)
 182                 continue
 183             filter_effect = 0
 184             for filter in filters:
 185                 try:
 186                     args = gen_orm_filter(ocloud.DeploymentManager, filter)
 187                 except KeyError:
 188                     logger.warning(
 189                         'Subscription {} filter {} has wrong attribute '
 190                         'name or value. Ignore the filter.'.format(
 191                             sub_data['subscriptionId'],
 192                             sub_data['filter']))
 193                     continue
 194                 if len(args) == 0 and 'objectType' in filter:
 195                     filter_effect += 1
 196                     break
 197                 args.append(
 198                     ocloud.DeploymentManager.deploymentManagerId == data.id)
 199                 ret = uow.deployment_managers.list_with_count(*args)
 200                 if ret[0] > 0:
 201                     logger.debug(
 202                         'DeploymentManager {} skip for subscription {} because'
 203                         ' of the filter.'
 204                         .format(data.id, sub_data['subscriptionId']))
 205                     filter_effect += 1
 206                     break
 207             if filter_effect > 0:
 208                 continue
 209             callback_smo(sub, data, dms_dict)
 210
 211
 212 def _notify_resource(uow, data):
 213     with uow:
 214         resource = uow.resources.get(data.id)
 215         if resource is None:
 216             logger.warning('Resource {} does not exists.'.format(data.id))
 217             return
 218         res_pool_id = resource.serialize()['resourcePoolId']
 219         logger.debug('res pool id is {}'.format(res_pool_id))
 220         res_dict = {
 221             'resourceId': resource.resourceId,
 222             'description': resource.description,
 223             'resourceTypeId': resource.resourceTypeId,
 224             'resourcePoolId': resource.resourcePoolId,
 225             'globalAssetId': resource.globalAssetId
 226         }
 227
 228         subs = uow.subscriptions.list()
 229         for sub in subs:
 230             sub_data = sub.serialize()
 231             logger.debug('Subscription: {}'.format(sub_data['subscriptionId']))
 232             filters = handle_filter(sub_data['filter'], 'ResourceInfo')
 233             if not filters:
 234                 callback_smo(sub, data, res_dict)
 235                 continue
 236             filter_effect = 0
 237             for filter in filters:
 238                 try:
 239                     args = gen_orm_filter(ocloud.Resource, filter)
 240                 except KeyError:
 241                     logger.warning(
 242                         'Subscription {} filter {} has wrong attribute '
 243                         'name or value. Ignore the filter.'.format(
 244                             sub_data['subscriptionId'],
 245                             sub_data['filter']))
 246                     continue
 247                 if len(args) == 0 and 'objectType' in filter:
 248                     filter_effect += 1
 249                     break
 250                 args.append(ocloud.Resource.resourceId == data.id)
 251                 ret = uow.resources.list_with_count(res_pool_id, *args)
 252                 if ret[0] > 0:
 253                     logger.debug(
 254                         'Resource {} skip for subscription {} because of '
 255                         'the filter.'
 256                         .format(data.id, sub_data['subscriptionId']))
 257                     filter_effect += 1
 258                     break
 259             if filter_effect > 0:
 260                 continue
 261             callback_smo(sub, data, res_dict)
 262
 263
 264 def handle_filter(filter: str, f_type: str):
 265     if not filter:
 266         return
 267     filter_strip = filter.strip(' []')
 268     filter_list = filter_strip.split('|')
 269     filters = list()
 270     for sub_filter in filter_list:
 271         exprs = sub_filter.split(';')
 272         objectType = False
 273         objectTypeValue = ''
 274         for expr in exprs:
 275             expr_strip = expr.strip(' ()')
 276             items = expr_strip.split(',')
 277             item_key = items[1].strip()
 278             if item_key != 'objectType':
 279                 continue
 280             objectType = True
 281             objectTypeValue = items[2].strip()
 282         if not objectType:
 283             if f_type == 'ResourceInfo':
 284                 filters.append(sub_filter)
 285             continue
 286         if objectTypeValue == f_type:
 287             filters.append(sub_filter)
 288     return filters
 289
 290
 291 def callback_smo(sub: Subscription, msg: Message2SMO, obj_dict: dict = None):
 292     sub_data = sub.serialize()
 293     callback = {
 294         'consumerSubscriptionId': sub_data['consumerSubscriptionId'],
 295         'notificationEventType': msg.notificationEventType,
 296         'objectRef': msg.objectRef,
 297         'updateTime': msg.updatetime
 298     }
 299     if msg.notificationEventType in [NotificationEventEnum.DELETE,
 300                                      NotificationEventEnum.MODIFY]:
 301         callback['priorObjectState'] = json.dumps(obj_dict)
 302     if msg.notificationEventType in [NotificationEventEnum.CREATE,
 303                                      NotificationEventEnum.MODIFY]:
 304         callback['postObjectState'] = json.dumps(obj_dict)
 305     if msg.notificationEventType == NotificationEventEnum.DELETE:
 306         callback.pop('objectRef')
 307     callback_data = json.dumps(callback)
 308     logger.info('URL: {}'.format(sub_data['callback']))
 309     logger.debug('callback data: {}'.format(callback_data))
 310
 311     # Call SMO through the SMO callback url
 312     o = urlparse(sub_data['callback'])
 313     if o.scheme == 'https':
 314         conn = get_https_conn_default(o.netloc)
 315     else:
 316         conn = get_http_conn(o.netloc)
 317     try:
 318         rst, status = post_data(conn, o.path, callback_data)
 319         if rst is True:
 320             logger.info(
 321                 'Notify to SMO successed with status: {}'.format(status))
 322             return
 323         logger.error('Notify Response code is: {}'.format(status))
 324     except ssl.SSLCertVerificationError as e:
 325         logger.debug(
 326             'Notify try to post data with trusted ca failed: {}'.format(e))
 327         if 'self signed' in str(e):
 328             conn = get_https_conn_selfsigned(o.netloc)
 329             try:
 330                 return post_data(conn, o.path, callback_data)
 331             except Exception as e:
 332                 logger.info(
 333                     'Notify post data with self-signed ca \
 334                     failed: {}'.format(e))
 335                 # TODO: write the status to extension db table.
 336                 return False
 337         return False
 338     except Exception as e:
 339         logger.critical('Notify except: {}'.format(e))
 340         return False