o2ims/service/command/registration_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 ssl
  16 import json
  17 from retry import retry
  18 from urllib.parse import urlparse
  19
  20 from o2common.config import config, conf
  21 from o2common.domain.filter import gen_orm_filter
  22 from o2common.service.unit_of_work import AbstractUnitOfWork
  23 from o2common.service.command.handler import get_https_conn_default, \
  24     get_http_conn, get_https_conn_selfsigned, post_data
  25
  26 from o2ims.domain import commands, ocloud as cloud
  27 from o2ims.domain.subscription_obj import Message2SMO, NotificationEventEnum
  28
  29 from .notify_handler import handle_filter, callback_smo
  30
  31 from o2common.helper import o2logging
  32 logger = o2logging.get_logger(__name__)
  33
  34 apibase = config.get_o2ims_api_base()
  35 api_monitoring_base = config.get_o2ims_monitoring_api_base()
  36 inventory_api_version = config.get_o2ims_inventory_api_v1()
  37
  38
  39 def registry_to_smo(
  40     cmd: commands.Register2SMO,
  41     uow: AbstractUnitOfWork,
  42 ):
  43     logger.debug('In registry_to_smo')
  44     data = cmd.data
  45     logger.info('The Register2SMO notificationEventType is {}'.format(
  46         data.notificationEventType))
  47     with uow:
  48         ocloud = uow.oclouds.get(data.id)
  49         if ocloud is None:
  50             logger.warning('Ocloud {} does not exists.'.format(data.id))
  51             return
  52         logger.debug('O-Cloud Global UUID: {}'.format(ocloud.globalCloudId))
  53         # ocloud_dict = ocloud.serialize()
  54         ocloud_dict = {
  55             'oCloudId': ocloud.oCloudId,
  56             'globalcloudId': ocloud.globalCloudId,
  57             'globalCloudId': ocloud.globalCloudId,
  58             'name': ocloud.name,
  59             'description': ocloud.description,
  60             'serviceUri': ocloud.serviceUri
  61         }
  62         if data.notificationEventType == NotificationEventEnum.CREATE:
  63             register_smo(uow, ocloud_dict)
  64         elif data.notificationEventType in [NotificationEventEnum.MODIFY,
  65                                             NotificationEventEnum.DELETE]:
  66             _notify_ocloud(uow, data, ocloud_dict)
  67
  68
  69 class RegIMSToSMOExp(Exception):
  70     def __init__(self, value):
  71         self.value = value
  72
  73
  74 def register_smo(uow, ocloud_data):
  75     call_res, status = call_smo(ocloud_data)
  76     logger.debug('Call SMO response is {}'.format(call_res))
  77     if call_res is True:
  78         logger.info('Register to smo success response is {}'.format(status))
  79     else:
  80         raise RegIMSToSMOExp('Register o2ims to SMO failed')
  81     # TODO: record the result for the smo register
  82
  83
  84 def _notify_ocloud(uow, data, ocloud_dict):
  85     ref = api_monitoring_base + inventory_api_version
  86     msg = Message2SMO(
  87         eventtype=data.notificationEventType, id=data.id,
  88         ref=ref, updatetime=data.updatetime)
  89     ocloud_dict.pop('globalCloudId')
  90     subs = uow.subscriptions.list()
  91     for sub in subs:
  92         sub_data = sub.serialize()
  93         logger.debug('Subscription: {}'.format(sub_data['subscriptionId']))
  94         filters = handle_filter(sub_data['filter'], 'CloudInfo')
  95         if not filters:
  96             callback_smo(sub, msg, ocloud_dict)
  97             continue
  98         filter_hit = False
  99         for filter in filters:
 100             try:
 101                 args = gen_orm_filter(cloud.Ocloud, filter)
 102             except KeyError:
 103                 logger.warning(
 104                     'Subscription {} filter {} has wrong attribute '
 105                     'name or value. Ignore the filter.'.format(
 106                         sub_data['subscriptionId'],
 107                         sub_data['filter']))
 108                 continue
 109             if len(args) == 0 and 'objectType' in filter:
 110                 filter_hit = True
 111                 break
 112             args.append(cloud.Ocloud.oCloudId == data.id)
 113             ret = uow.oclouds.list(*args)
 114             if ret.count() > 0:
 115                 filter_hit = True
 116                 break
 117         if filter_hit:
 118             logger.info('Subscription {} filter hit, skip oCloud {}.'
 119                         .format(sub_data['subscriptionId'], data.id))
 120         else:
 121             callback_smo(sub, msg, ocloud_dict)
 122
 123
 124 @retry((ConnectionRefusedError), tries=2, delay=2)
 125 def call_smo(reg_data: dict):
 126     smo_token = conf.DEFAULT.smo_token_data
 127     smo_token_info = {
 128         'iss': 'o2ims',
 129         'aud': 'smo',
 130         'smo_token_payload': smo_token,
 131         'smo_token_type': 'jwt',
 132         'smo_token_expiration': '',
 133         'smo_token_algo': 'RS256'
 134     }
 135
 136     callback_data = json.dumps({
 137         'globalCloudId': reg_data['globalCloudId'],
 138         'oCloudId': reg_data['oCloudId'],
 139         'IMS_EP': config.get_api_url(),
 140         'smo_token_data': smo_token_info
 141     })
 142     logger.info('callback URL: {}'.format(conf.DEFAULT.smo_register_url))
 143     logger.debug('callback data: {}'.format(callback_data))
 144     o = urlparse(conf.DEFAULT.smo_register_url)
 145     if o.scheme == 'https':
 146         conn = get_https_conn_default(o.netloc)
 147     else:
 148         conn = get_http_conn(o.netloc)
 149
 150     try:
 151         return post_data(conn, o.path, callback_data)
 152     except ssl.SSLCertVerificationError as e:
 153         logger.debug('Try to register to smo with \
 154         trusted ca failed: {}'.format(e))
 155         if 'self signed' in str(e):
 156             conn = get_https_conn_selfsigned(o.netloc)
 157             try:
 158                 return post_data(conn, o.path, callback_data)
 159             except Exception as e:
 160                 logger.info(
 161                     'Register to smo with self-signed ca failed: {}'.format(e))
 162                 # TODO: write the status to extension db table.
 163                 return False, None
 164         return False, None
 165     except Exception as e:
 166         logger.critical('Register to smo except: {}'.format(e))
 167         return False, None