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