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_hit = False
  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_hit = True
  94                     break
  95                 args.append(ocloud.ResourceType.resourceTypeId == data.id)
  96                 obj_count, _ = uow.resource_types.list_with_count(*args)
  97                 if obj_count > 0:
  98                     filter_hit = True
  99                     break
 100             if filter_hit:
 101                 logger.info('Subscription {} filter hit, skip ResourceType {}.'
 102                             .format(sub_data['subscriptionId'], data.id))
 103             else:
 104                 callback_smo(sub, data, resource_type_dict)
 105
 106
 107 def _notify_resourcepool(uow, data):
 108     with uow:
 109         resource_pool = uow.resource_pools.get(data.id)
 110         if resource_pool is None:
 111             logger.warning('ResourcePool {} does not exists.'.format(data.id))
 112             return
 113         resource_pool_dict = {
 114             'resourcePoolId': resource_pool.resourcePoolId,
 115             'oCloudId': resource_pool.oCloudId,
 116             'globalLocationId': resource_pool.globalLocationId,
 117             'name': resource_pool.name,
 118             'description': resource_pool.description
 119         }
 120
 121         subs = uow.subscriptions.list()
 122         for sub in subs:
 123             sub_data = sub.serialize()
 124             logger.debug('Subscription: {}'.format(sub_data['subscriptionId']))
 125             filters = handle_filter(sub_data['filter'], 'ResourcePoolInfo')
 126             if not filters:
 127                 callback_smo(sub, data, resource_pool_dict)
 128                 continue
 129             filter_hit = False
 130             for filter in filters:
 131                 try:
 132                     args = gen_orm_filter(ocloud.ResourcePool, filter)
 133                 except KeyError:
 134                     logger.warning(
 135                         'Subscription {} filter {} has wrong attribute '
 136                         'name or value. Ignore the filter.'.format(
 137                             sub_data['subscriptionId'],
 138                             sub_data['filter']))
 139                     continue
 140                 if len(args) == 0 and 'objectType' in filter:
 141                     filter_hit = True
 142                     break
 143                 args.append(ocloud.ResourcePool.resourcePoolId == data.id)
 144                 obj_count, _ = uow.resource_pools.list_with_count(*args)
 145                 if obj_count > 0:
 146                     filter_hit = True
 147                     break
 148             if filter_hit:
 149                 logger.info('Subscription {} filter hit, skip ResourcePool {}.'
 150                             .format(sub_data['subscriptionId'], data.id))
 151             else:
 152                 callback_smo(sub, data, resource_pool_dict)
 153
 154
 155 def _notify_dms(uow, data):
 156     with uow:
 157         dms = uow.deployment_managers.get(data.id)
 158         if dms is None:
 159             logger.warning(
 160                 'DeploymentManager {} does not exists.'.format(data.id))
 161             return
 162         dms_dict = {
 163             'deploymentManagerId': dms.deploymentManagerId,
 164             'name': dms.name,
 165             'description': dms.description,
 166             'oCloudId': dms.oCloudId,
 167             'serviceUri': dms.serviceUri
 168         }
 169
 170         subs = uow.subscriptions.list()
 171         for sub in subs:
 172             sub_data = sub.serialize()
 173             logger.debug('Subscription: {}'.format(sub_data['subscriptionId']))
 174             filters = handle_filter(
 175                 sub_data['filter'], 'DeploymentManagerInfo')
 176             if not filters:
 177                 callback_smo(sub, data, dms_dict)
 178                 continue
 179             filter_hit = False
 180             for filter in filters:
 181                 try:
 182                     args = gen_orm_filter(ocloud.DeploymentManager, filter)
 183                 except KeyError:
 184                     logger.warning(
 185                         'Subscription {} filter {} has wrong attribute '
 186                         'name or value. Ignore the filter.'.format(
 187                             sub_data['subscriptionId'],
 188                             sub_data['filter']))
 189                     continue
 190                 if len(args) == 0 and 'objectType' in filter:
 191                     filter_hit = True
 192                     break
 193                 args.append(
 194                     ocloud.DeploymentManager.deploymentManagerId == data.id)
 195                 obj_count, _ = uow.deployment_managers.list_with_count(*args)
 196                 if obj_count > 0:
 197                     filter_hit = True
 198                     break
 199             if filter_hit:
 200                 logger.info('Subscription {} filter hit, skip '
 201                             'DeploymentManager {}.'
 202                             .format(sub_data['subscriptionId'], data.id))
 203             else:
 204                 callback_smo(sub, data, dms_dict)
 205
 206
 207 def _notify_resource(uow, data):
 208     with uow:
 209         resource = uow.resources.get(data.id)
 210         if resource is None:
 211             logger.warning('Resource {} does not exists.'.format(data.id))
 212             return
 213         res_pool_id = resource.serialize()['resourcePoolId']
 214         logger.debug('res pool id is {}'.format(res_pool_id))
 215         res_dict = {
 216             'resourceId': resource.resourceId,
 217             'description': resource.description,
 218             'resourceTypeId': resource.resourceTypeId,
 219             'resourcePoolId': resource.resourcePoolId,
 220             'globalAssetId': resource.globalAssetId
 221         }
 222
 223         subs = uow.subscriptions.list()
 224         for sub in subs:
 225             sub_data = sub.serialize()
 226             logger.debug('Subscription: {}'.format(sub_data['subscriptionId']))
 227             filters = handle_filter(sub_data['filter'], 'ResourceInfo')
 228             if not filters:
 229                 callback_smo(sub, data, res_dict)
 230                 continue
 231             filter_hit = False
 232             for filter in filters:
 233                 try:
 234                     args = gen_orm_filter(ocloud.Resource, filter)
 235                 except KeyError:
 236                     logger.warning(
 237                         'Subscription {} filter {} has wrong attribute '
 238                         'name or value. Ignore the filter.'.format(
 239                             sub_data['subscriptionId'],
 240                             sub_data['filter']))
 241                     continue
 242                 if len(args) == 0 and 'objectType' in filter:
 243                     filter_hit = True
 244                     break
 245                 args.append(ocloud.Resource.resourceId == data.id)
 246                 obj_count, _ = uow.resources.list_with_count(
 247                     res_pool_id, *args)
 248                 if obj_count > 0:
 249                     filter_hit = True
 250                     break
 251             if filter_hit:
 252                 logger.info('Subscription {} filter hit, skip Resource {}.'
 253                             .format(sub_data['subscriptionId'], data.id))
 254             else:
 255                 callback_smo(sub, data, res_dict)
 256
 257
 258 def handle_filter(filter: str, f_type: str):
 259     if not filter:
 260         return
 261     filter_strip = filter.strip(' []')
 262     filter_list = filter_strip.split('|')
 263     filters = list()
 264     for sub_filter in filter_list:
 265         exprs = sub_filter.split(';')
 266         objectType = False
 267         objectTypeValue = ''
 268         for expr in exprs:
 269             expr_strip = expr.strip(' ()')
 270             items = expr_strip.split(',')
 271             item_key = items[1].strip()
 272             if item_key != 'objectType':
 273                 continue
 274             objectType = True
 275             objectTypeValue = items[2].strip()
 276         if not objectType:
 277             if f_type == 'ResourceInfo':
 278                 filters.append(sub_filter)
 279             continue
 280         if objectTypeValue == f_type:
 281             filters.append(sub_filter)
 282     return filters
 283
 284
 285 def callback_smo(sub: Subscription, msg: Message2SMO, obj_dict: dict = None):
 286     sub_data = sub.serialize()
 287     callback = {
 288         'consumerSubscriptionId': sub_data['consumerSubscriptionId'],
 289         'notificationEventType': msg.notificationEventType,
 290         'objectRef': msg.objectRef,
 291         'updateTime': msg.updatetime
 292     }
 293     if msg.notificationEventType in [NotificationEventEnum.DELETE,
 294                                      NotificationEventEnum.MODIFY]:
 295         callback['priorObjectState'] = json.dumps(obj_dict)
 296     if msg.notificationEventType in [NotificationEventEnum.CREATE,
 297                                      NotificationEventEnum.MODIFY]:
 298         callback['postObjectState'] = json.dumps(obj_dict)
 299     if msg.notificationEventType == NotificationEventEnum.DELETE:
 300         callback.pop('objectRef')
 301     callback_data = json.dumps(callback)
 302     logger.info('callback URL: {}'.format(sub_data['callback']))
 303     logger.debug('callback data: {}'.format(callback_data))
 304
 305     # Call SMO through the SMO callback url
 306     o = urlparse(sub_data['callback'])
 307     if o.scheme == 'https':
 308         conn = get_https_conn_default(o.netloc)
 309     else:
 310         conn = get_http_conn(o.netloc)
 311     try:
 312         rst, status = post_data(conn, o.path, callback_data)
 313         if rst is True:
 314             logger.info(
 315                 'Notify to SMO successed with status: {}'.format(status))
 316             return
 317         logger.error('Notify Response code is: {}'.format(status))
 318     except ssl.SSLCertVerificationError as e:
 319         logger.debug(
 320             'Notify try to post data with trusted ca failed: {}'.format(e))
 321         if 'self signed' in str(e):
 322             conn = get_https_conn_selfsigned(o.netloc)
 323             try:
 324                 return post_data(conn, o.path, callback_data)
 325             except Exception as e:
 326                 logger.info(
 327                     'Notify post data with self-signed ca \
 328                     failed: {}'.format(e))
 329                 # TODO: write the status to extension db table.
 330                 return False
 331         return False
 332     except Exception as e:
 333         logger.critical('Notify except: {}'.format(e))
 334         return False