lcm/lcm/pub/utils/notificationsutil.py

   1 # Copyright (C) 2018 Verizon. All Rights Reserved
   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 json
  16 import base64
  17 import sys
  18 import traceback
  19 import logging
  20 import urllib.request
  21 import urllib.error
  22 import urllib.parse
  23 import uuid
  24 import httplib2
  25
  26 from lcm.nf import const
  27 from lcm.pub.database.models import SubscriptionModel
  28 from lcm.pub.database.models import VmInstModel
  29 from lcm.pub.database.models import NetworkInstModel
  30 from lcm.pub.database.models import PortInstModel
  31 from lcm.pub.database.models import StorageInstModel
  32 from lcm.pub.database.models import VNFCInstModel
  33 from lcm.pub.database.models import NfInstModel
  34 from lcm.pub.utils.timeutil import now_time
  35 from lcm.pub.utils.enumutil import enum
  36
  37 logger = logging.getLogger(__name__)
  38
  39 rest_no_auth, rest_oneway_auth, rest_bothway_auth = 0, 1, 2
  40 HTTP_200_OK, HTTP_201_CREATED, HTTP_204_NO_CONTENT, HTTP_202_ACCEPTED = '200', '201', '204', '202'
  41 status_ok_list = [HTTP_200_OK, HTTP_201_CREATED, HTTP_204_NO_CONTENT, HTTP_202_ACCEPTED]
  42 HTTP_404_NOTFOUND, HTTP_403_FORBIDDEN, HTTP_401_UNAUTHORIZED, HTTP_400_BADREQUEST = '404', '403', '401', '400'
  43 NOTIFY_TYPE = enum(
  44     lCM_OP_OCC="VnfLcmOperationOccurrenceNotification",
  45     CREATION="VnfIdentifierCreationNotification",
  46     DELETION="VnfIdentifierDeletionNotification"
  47 )
  48
  49
  50 class NotificationsUtil(object):
  51     def send_notification(self, notification):
  52         logger.info("Send Notifications to the callbackUri")
  53         filters = {
  54             "operationState": "operation_states",
  55             "operation": "operation_types",
  56             "vnfInstanceId": "vnf_instance_filter"
  57         }
  58         subscriptions_filter = {v + "__contains": notification[k] for k, v in list(filters.items())}
  59
  60         subscriptions = SubscriptionModel.objects.filter(**subscriptions_filter)
  61         if not subscriptions.exists():
  62             logger.info("No subscriptions created for the filters %s" % notification)
  63             return
  64         logger.info("Start sending notifications")
  65         for subscription in subscriptions:
  66             # set subscription id
  67             notification["subscriptionId"] = subscription.subscription_id
  68             notification['_links']['subscription'] = {
  69                 'href': '/api/vnflcm/v1/subscriptions/%s' % subscription.subscription_id
  70             }
  71             callbackUri = subscription.callback_uri
  72             auth_info = json.loads(subscription.auth_info)
  73             if const.BASIC in auth_info["authType"]:
  74                 try:
  75                     self.post_notification(callbackUri, notification)
  76                 except Exception as e:
  77                     logger.error("Failed to post notification: %s", e.args[0])
  78
  79     def post_notification(self, callbackUri, notification):
  80         logger.info("Sending notification to %s", callbackUri)
  81         resp = self.call_req(callbackUri, "", "", "POST", json.dumps(notification))
  82         if resp[0] != 0:
  83             logger.error('Status code is %s, detail is %s.', resp[2], resp[1])
  84
  85     def call_req(self, full_url, user, passwd, method, content=''):
  86         callid = str(uuid.uuid1())
  87         logger.debug("[%s]call_req('%s','%s','%s',%s,'%s','%s')" % (
  88             callid, full_url, user, passwd, rest_no_auth, method, content))
  89         ret = None
  90         resp_Location = ''
  91         resp_status = ''
  92         try:
  93             headers = {'content-type': 'application/json', 'accept': 'application/json'}
  94             if user:
  95                 headers['Authorization'] = 'Basic %s' % base64.b64encode(
  96                     bytes('%s:%s' % (user, passwd), "utf-8")).decode()
  97             ca_certs = None
  98             for retry_times in range(3):
  99                 http = httplib2.Http(ca_certs=ca_certs, disable_ssl_certificate_validation=True)
 100                 http.follow_all_redirects = True
 101                 try:
 102                     resp, resp_content = http.request(full_url, method=method.upper(), body=content, headers=headers)
 103                     logger.info("resp=%s,resp_content=%s" % (resp, resp_content))
 104                     resp_status, resp_body = resp['status'], resp_content.decode('UTF-8')
 105                     resp_Location = resp.get('Location', "")
 106                     logger.debug("[%s][%d]status=%s,resp_body=%s)" % (callid, retry_times, resp_status, resp_body))
 107                     if resp_status in status_ok_list:
 108                         ret = [0, resp_body, resp_status, resp_Location]
 109                     else:
 110                         ret = [1, resp_body, resp_status, resp_Location]
 111                     break
 112                 except Exception as ex:
 113                     if 'httplib.ResponseNotReady' in str(sys.exc_info()):
 114                         logger.debug("retry_times=%d", retry_times)
 115                         logger.error(traceback.format_exc())
 116                         ret = [1, "Unable to connect to %s" % full_url, resp_status, resp_Location]
 117                         continue
 118                     raise ex
 119         except urllib.error.URLError as err:
 120             ret = [2, str(err), resp_status, resp_Location]
 121         except Exception:
 122             logger.error(traceback.format_exc())
 123             logger.error("[%s]ret=%s" % (callid, str(sys.exc_info())))
 124             res_info = str(sys.exc_info())
 125             if 'httplib.ResponseNotReady' in res_info:
 126                 res_info = "The URL[%s] request failed or is not responding." % full_url
 127             ret = [3, res_info, resp_status, resp_Location]
 128         except:
 129             logger.error(traceback.format_exc())
 130             ret = [4, str(sys.exc_info()), resp_status, resp_Location]
 131
 132         logger.debug("[%s]ret=%s" % (callid, str(ret)))
 133         return ret
 134
 135
 136 def set_affected_vnfcs(affected_vnfcs, nfinstid, changetype):
 137     vnfcs = VNFCInstModel.objects.filter(instid=nfinstid)
 138     for vnfc in vnfcs:
 139         vm_resource = {}
 140         if vnfc.vmid:
 141             vm = VmInstModel.objects.filter(vmid=vnfc.vmid)
 142             if vm:
 143                 vm_resource = {
 144                     'vimConnectionId': vm[0].vimid,
 145                     'resourceId': vm[0].resourceid,
 146                     'resourceProviderId': vm[0].vmname,  # TODO: is resourceName mapped to resourceProviderId?
 147                     'vimLevelResourceType': 'vm'
 148                 }
 149         affected_vnfcs.append({
 150             'id': vnfc.vnfcinstanceid,
 151             'vduId': vnfc.vduid,
 152             'changeType': changetype,
 153             'computeResource': vm_resource
 154         })
 155     logger.debug("affected_vnfcs=%s", affected_vnfcs)
 156     return affected_vnfcs
 157
 158
 159 def set_affected_vls(affected_vls, nfinstid, changetype):
 160     networks = NetworkInstModel.objects.filter(instid=nfinstid)
 161     for network in networks:
 162         network_resource = {
 163             'vimConnectionId': network.vimid,
 164             'resourceId': network.resourceid,
 165             'resourceProviderId': network.name,  # TODO: is resourceName mapped to resourceProviderId?
 166             'vimLevelResourceType': 'network'
 167         }
 168         affected_vls.append({
 169             'id': network.networkid,
 170             'virtualLinkDescId': network.nodeId,
 171             'changeType': changetype,
 172             'networkResource': network_resource
 173         })
 174     logger.debug("affected_vls=%s", affected_vls)
 175
 176
 177 def set_ext_connectivity(ext_connectivity, nfinstid):
 178     ext_connectivity_map = {}
 179     ports = PortInstModel.objects.filter(instid=nfinstid)
 180     for port in ports:
 181         if port.networkid not in ext_connectivity_map:
 182             ext_connectivity_map[port.networkid] = []
 183         ext_connectivity_map[port.networkid].append({
 184             'id': port.portid,  # TODO: port.portid or port.nodeid?
 185             'resourceHandle': {
 186                 'vimConnectionId': port.vimid,
 187                 'resourceId': port.resourceid,
 188                 'resourceProviderId': port.name,  # TODO: is resourceName mapped to resourceProviderId?
 189                 'vimLevelResourceType': 'port'
 190             },
 191             'cpInstanceId': port.portid  # TODO: port.cpinstanceid is not initiated when create port resource.
 192         })
 193     for network_id, ext_link_ports in list(ext_connectivity_map.items()):
 194         networks = NetworkInstModel.objects.filter(networkid=network_id)
 195         net_name = networks[0].name if networks else network_id
 196         network_resource = {
 197             'vimConnectionId': ext_link_ports[0]['resourceHandle']['vimConnectionId'],
 198             'resourceId': network_id,
 199             'resourceProviderId': net_name,  # TODO: is resourceName mapped to resourceProviderId?
 200             'vimLevelResourceType': 'network'
 201         }
 202         ext_connectivity.append({
 203             'id': network_id,
 204             'resourceHandle': network_resource,
 205             'extLinkPorts': ext_link_ports
 206         })
 207     logger.debug("ext_connectivity=%s", ext_connectivity)
 208
 209
 210 def set_affected_vss(affected_vss, nfinstid, changetype):
 211     vss = StorageInstModel.objects.filter(instid=nfinstid)
 212     for vs in vss:
 213         affected_vss.append({
 214             'id': vs.storageid,
 215             'virtualStorageDescId': vs.nodeId,
 216             'changeType': changetype,
 217             'storageResource': {
 218                 'vimConnectionId': vs.vimid,
 219                 'resourceId': vs.resourceid,
 220                 'resourceProviderId': vs.name,  # TODO: is resourceName mapped to resourceProviderId?
 221                 'vimLevelResourceType': 'volume'
 222             }
 223         })
 224     logger.debug("affected_vss=%s", affected_vss)
 225
 226
 227 def get_notification_status(operation_state):
 228     if operation_state == const.OPERATION_STATE_TYPE.STARTING:
 229         return const.LCM_NOTIFICATION_STATUS.START
 230     return const.LCM_NOTIFICATION_STATUS.RESULT
 231
 232
 233 def prepare_notification(nfinstid, jobid, operation, operation_state):
 234     logger.info('Start to prepare notification')
 235     notification_content = {
 236         'id': str(uuid.uuid4()),  # shall be the same if sent multiple times due to multiple subscriptions.
 237         'notificationType': NOTIFY_TYPE.lCM_OP_OCC,
 238         # set 'subscriptionId' after filtering for subscribers
 239         'timeStamp': now_time(),
 240         'notificationStatus': get_notification_status(operation_state),
 241         'operationState': operation_state,
 242         'vnfInstanceId': nfinstid,
 243         'operation': operation,
 244         'isAutomaticInvocation': False,
 245         'vnfLcmOpOccId': jobid,
 246         'affectedVnfcs': [],
 247         'affectedVirtualLinks': [],
 248         'affectedVirtualStorages': [],
 249         'changedExtConnectivity': [],
 250         'error': None,
 251         '_links': {
 252             'vnfInstance': {
 253                 'href': '%s/vnf_instances/%s' % (const.URL_PREFIX, nfinstid)
 254             },
 255             'vnfLcmOpOcc': {
 256                 'href': '%s/vnf_lcm_op_occs/%s' % (const.URL_PREFIX, jobid)
 257             }
 258         }
 259     }
 260     nfInsts = NfInstModel.objects.filter(nfinstid=nfinstid)
 261     if nfInsts.exists():
 262         notification_content['vnfmInstId'] = nfInsts[0].vnfminstid
 263     else:
 264         notification_content['vnfmInstId'] = "1"
 265     return notification_content
 266
 267
 268 def prepare_notification_data(nfinstid, jobid, changetype, operation):
 269     data = prepare_notification(
 270         nfinstid=nfinstid,
 271         jobid=jobid,
 272         operation=operation,
 273         operation_state=const.OPERATION_STATE_TYPE.COMPLETED
 274     )
 275
 276     set_affected_vnfcs(data['affectedVnfcs'], nfinstid, changetype)
 277     set_affected_vls(data['affectedVirtualLinks'], nfinstid, changetype)
 278     set_affected_vss(data['affectedVirtualStorages'], nfinstid, changetype)
 279     set_ext_connectivity(data['changedExtConnectivity'], nfinstid)
 280
 281     logger.debug('Notification content: %s' % data)
 282     return data
 283
 284
 285 def prepare_vnf_identifier_notification(notify_type, nfinstid):
 286     data = {
 287         "id": str(uuid.uuid4()),  # shall be the same if sent multiple times due to multiple subscriptions.
 288         "notificationType": notify_type,
 289         "timeStamp": now_time(),
 290         "vnfInstanceId": nfinstid,
 291         "_links": {
 292             "vnfInstance": {
 293                 'href': '%s/vnf_instances/%s' % (const.URL_PREFIX, nfinstid)
 294             }
 295         }
 296     }
 297
 298     logger.debug('Vnf Identifier Notification: %s' % data)
 299     return data