-# Copyright (C) 2021 Wind River Systems, Inc.
+# Copyright (C) 2021-2022 Wind River Systems, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# See the License for the specific language governing permissions and
# limitations under the License.
-# import time
import json
-# import asyncio
-# import requests
-import http.client
-import ssl
-from urllib.parse import urlparse
-from retry import retry
-from o2common.service.unit_of_work import AbstractUnitOfWork
from o2common.config import config, conf
+from o2common.domain.filter import gen_orm_filter
+from o2common.service.unit_of_work import AbstractUnitOfWork
+from o2common.adapter.notifications import AbstractNotifications
-from o2ims.domain import commands
-from o2ims.domain.subscription_obj import NotificationEventEnum
+from o2ims.domain import commands, ocloud as cloud
+from o2ims.domain.subscription_obj import Message2SMO, NotificationEventEnum
+
+from .notify_handler import handle_filter, callback_smo
from o2common.helper import o2logging
logger = o2logging.get_logger(__name__)
+apibase = config.get_o2ims_api_base()
+api_monitoring_base = config.get_o2ims_monitoring_api_base()
+inventory_api_version = config.get_o2ims_inventory_api_v1()
+
def registry_to_smo(
cmd: commands.Register2SMO,
uow: AbstractUnitOfWork,
+ notifications: AbstractNotifications,
):
- logger.info('In registry_to_smo')
+ logger.debug('In registry_to_smo')
data = cmd.data
logger.info('The Register2SMO notificationEventType is {}'.format(
data.notificationEventType))
with uow:
ocloud = uow.oclouds.get(data.id)
if ocloud is None:
+ logger.warning('Ocloud {} does not exists.'.format(data.id))
return
logger.debug('O-Cloud Global UUID: {}'.format(ocloud.globalCloudId))
- ocloud_dict = ocloud.serialize()
+ # ocloud_dict = ocloud.serialize()
+ ocloud_dict = {
+ 'oCloudId': ocloud.oCloudId,
+ 'globalcloudId': ocloud.globalCloudId,
+ 'globalCloudId': ocloud.globalCloudId,
+ 'name': ocloud.name,
+ 'description': ocloud.description,
+ 'serviceUri': ocloud.serviceUri
+ }
if data.notificationEventType == NotificationEventEnum.CREATE:
- register_smo(uow, ocloud_dict)
+ register_smo(notifications, ocloud_dict)
+ elif data.notificationEventType in [NotificationEventEnum.MODIFY,
+ NotificationEventEnum.DELETE]:
+ _notify_ocloud(uow, data, ocloud_dict)
class RegIMSToSMOExp(Exception):
self.value = value
-def register_smo(uow, ocloud_data):
- call_res = call_smo(ocloud_data)
+def register_smo(notifications, ocloud_data):
+ call_res = call_smo(notifications, ocloud_data)
logger.debug('Call SMO response is {}'.format(call_res))
- if call_res is not True:
+ if call_res is True:
+ logger.info('Register to smo success response')
+ else:
raise RegIMSToSMOExp('Register o2ims to SMO failed')
# TODO: record the result for the smo register
-# def retry(fun, max_tries=2):
-# for i in range(max_tries):
-# try:
-# time.sleep(5*i)
-# # await asyncio.sleep(5*i)
-# res = fun()
-# logger.debug('retry function result: {}'.format(res))
-# return res
-# except Exception:
-# continue
-
-
-@retry((ConnectionRefusedError), tries=2, delay=2)
-def call_smo(reg_data: dict):
+def _notify_ocloud(uow, data, ocloud_dict):
+ ref = api_monitoring_base + inventory_api_version
+ msg = Message2SMO(
+ eventtype=data.notificationEventType, id=data.id,
+ ref=ref, updatetime=data.updatetime)
+ ocloud_dict.pop('globalCloudId')
+ subs = uow.subscriptions.list()
+ for sub in subs:
+ sub_data = sub.serialize()
+ logger.debug('Subscription: {}'.format(sub_data['subscriptionId']))
+ filters = handle_filter(sub_data['filter'], 'CloudInfo')
+ if not filters:
+ callback_smo(sub, msg, ocloud_dict)
+ continue
+ filter_hit = False
+ for filter in filters:
+ try:
+ args = gen_orm_filter(cloud.Ocloud, filter)
+ except KeyError:
+ logger.warning(
+ 'Subscription {} filter {} has wrong attribute '
+ 'name or value. Ignore the filter.'.format(
+ sub_data['subscriptionId'],
+ sub_data['filter']))
+ continue
+ if len(args) == 0 and 'objectType' in filter:
+ filter_hit = True
+ break
+ args.append(cloud.Ocloud.oCloudId == data.id)
+ ret = uow.oclouds.list(*args)
+ if ret.count() > 0:
+ filter_hit = True
+ break
+ if filter_hit:
+ logger.info('Subscription {} filter hit, skip oCloud {}.'
+ .format(sub_data['subscriptionId'], data.id))
+ else:
+ callback_smo(sub, msg, ocloud_dict)
+
+
+def call_smo(notifications: AbstractNotifications, reg_data: dict):
smo_token = conf.DEFAULT.smo_token_data
smo_token_info = {
'iss': 'o2ims',
'IMS_EP': config.get_api_url(),
'smo_token_data': smo_token_info
})
- logger.info('URL: {}, data: {}'.format(
- conf.DEFAULT.smo_register_url, callback_data))
- o = urlparse(conf.DEFAULT.smo_register_url)
- if o.scheme == 'https':
- sslctx = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH)
- sslctx.check_hostname = True
- sslctx.verify_mode = ssl.CERT_REQUIRED
- sslctx.load_default_certs()
- conn = http.client.HTTPSConnection(o.netloc, context=sslctx)
- else:
- conn = http.client.HTTPConnection(o.netloc)
-
- try:
- return post_data(conn, o.path, callback_data)
- except ssl.SSLCertVerificationError as e:
- logger.info('post data except: {}'.format(e))
- if 'self signed' in str(e):
- sslctx = ssl.create_default_context(
- purpose=ssl.Purpose.SERVER_AUTH)
- smo_ca_path = config.get_smo_ca_config_path()
- sslctx.load_verify_locations(smo_ca_path)
- sslctx.check_hostname = False
- sslctx.verify_mode = ssl.CERT_REQUIRED
- conn = http.client.HTTPSConnection(o.netloc, context=sslctx)
- return post_data(conn, o.path, callback_data)
- except Exception as e:
- logger.info('except: {}'.format(e))
- return False
-
-
-def post_data(conn, path, data):
- headers = {'Content-type': 'application/json'}
- conn.request('POST', path, data, headers)
- resp = conn.getresponse()
- data = resp.read().decode('utf-8')
- # json_data = json.loads(data)
- if resp.status == 202 or resp.status == 200:
- logger.info('Registrer to SMO successed, response code {} {}, data {}'.
- format(resp.status, resp.reason, data))
- return True
- logger.error('Response code is: {}'.format(resp.status))
- return False
+ logger.info('callback URL: {}'.format(conf.DEFAULT.smo_register_url))
+ logger.debug('callback data: {}'.format(callback_data))
+ return notifications.send(conf.DEFAULT.smo_register_url, callback_data)
Code Review / pti / o2.git / blobdiff