lcm/lcm/nf/biz/scale_vnf.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 import json
  15 import logging
  16 import uuid
  17 import traceback
  18 from threading import Thread
  19
  20 from lcm.nf.biz import common
  21 from lcm.nf.biz.grant_vnf import grant_resource
  22 from lcm.nf.const import GRANT_TYPE, CHANGE_TYPE
  23 from lcm.nf.const import RESOURCE_MAP, OPERATION_STATE_TYPE
  24 from lcm.nf.const import INSTANTIATION_STATE
  25 from lcm.nf.const import OPERATION_TYPE
  26 from lcm.nf.const import OPERATION_TASK
  27 from lcm.pub.database.models import NfInstModel
  28 from lcm.pub.database.models import VNFCInstModel, PortInstModel
  29 from lcm.pub.database.models import VmInstModel
  30 from lcm.pub.exceptions import NFLCMException
  31 from lcm.pub.utils.notificationsutil import NotificationsUtil, prepare_notification
  32 from lcm.pub.utils.values import ignore_case_get
  33 from lcm.pub.utils.jobutil import JobUtil
  34 from lcm.pub.utils.timeutil import now_time
  35 from lcm.pub.vimapi import adaptor
  36 from .operate_vnf_lcm_op_occ import VnfLcmOpOcc
  37
  38 logger = logging.getLogger(__name__)
  39
  40 DEFAULT_STEPS = 1
  41
  42
  43 class ScaleVnf(Thread):
  44     def __init__(self, data, nf_inst_id, job_id):
  45         super(ScaleVnf, self).__init__()
  46         self.data = data
  47         self.nf_inst_id = nf_inst_id
  48         self.job_id = job_id
  49         self.vnf_insts = NfInstModel.objects.filter(nfinstid=self.nf_inst_id)
  50         self.lcm_op_occ = VnfLcmOpOcc(
  51             vnf_inst_id=nf_inst_id,
  52             lcm_op_id=job_id,
  53             operation=OPERATION_TYPE.SCALE,
  54             task=OPERATION_TASK.SCALE
  55         )
  56         self.op_type = OPERATION_TYPE.SCALE
  57
  58     def run(self):
  59         try:
  60             self.scale_pre()
  61             self.lcm_op_occ.notify_lcm(OPERATION_STATE_TYPE.STARTING)
  62             JobUtil.add_job_status(self.job_id,
  63                                    50,
  64                                    "Start to apply grant.")
  65             self.apply_grant()
  66             self.lcm_op_occ.notify_lcm(OPERATION_STATE_TYPE.PROCESSING)
  67             JobUtil.add_job_status(self.job_id,
  68                                    75,
  69                                    "Start to scale Vnf.")
  70             self.do_operation()
  71             self.send_notification()
  72             JobUtil.add_job_status(self.job_id,
  73                                    100,
  74                                    "Scale Vnf success.")
  75             self.vnf_insts.update(status=INSTANTIATION_STATE.INSTANTIATED,
  76                                   lastuptime=now_time())
  77         except NFLCMException as e:
  78             logger.error(e.message)
  79             self.vnf_scale_failed_handle(e.message)
  80         except Exception as e:
  81             logger.error(e.message)
  82             logger.error(traceback.format_exc())
  83             self.vnf_scale_failed_handle(e.message)
  84
  85     def scale_pre(self):
  86         self.scale_type = self.data.get("type")
  87         self.aspect_id = self.data.get("aspectId")
  88         self.number_of_steps = int(self.data.get("numberOfSteps", DEFAULT_STEPS))
  89         self.additional_params = self.data.get("additionalParams", {})
  90         self.is_scale_in = (self.scale_type == GRANT_TYPE.SCALE_IN)
  91         self.vnfd_info = json.loads(self.vnf_insts[0].vnfd_model)
  92         self.step_delta = self.get_scale_step_delta()
  93         self.target_vdu, self.step_inst_num = self.get_vdu_scale_aspect_deltas()
  94         self.scale_inst_num = self.number_of_steps * self.step_inst_num
  95         self.min_instance_num, self.max_instance_num = self.get_instance_range()
  96         self.check_if_can_scale()
  97         self.scale_out_resource = {}
  98
  99     def apply_grant(self):
 100         logger.debug("Start scale apply grant")
 101         vdus = ignore_case_get(self.vnfd_info, "vdus")
 102         scale_vdus = [vdu for vdu in vdus if vdu["vdu_id"] == self.target_vdu]
 103         scale_vdus = scale_vdus * self.scale_inst_num
 104         grant_result = grant_resource(data=self.data,
 105                                       nf_inst_id=self.nf_inst_id,
 106                                       job_id=self.job_id,
 107                                       grant_type=self.scale_type,
 108                                       vdus=scale_vdus)
 109         logger.debug("Scale Grant result: %s", grant_result)
 110         self.set_location(grant_result)
 111
 112     def do_operation(self):
 113         logger.debug("Start %s VNF resource", self.scale_type)
 114         logger.debug('VnfdInfo = %s' % self.vnfd_info)
 115
 116         if self.is_scale_in:
 117             self.affected_vnfcs = []
 118             self.scale_in_resource = self.get_scale_in_resource(self.affected_vnfcs)
 119             adaptor.delete_vim_res(self.scale_in_resource, self.do_notify_del_vim_res)
 120         else:
 121             self.scale_out_resource = {
 122                 'volumn': [],
 123                 'network': [],
 124                 'subnet': [],
 125                 'port': [],
 126                 'flavor': [],
 127                 'vm': []
 128             }
 129             self.vnfd_info["volume_storages"] = []
 130             self.vnfd_info["vls"] = []
 131             self.vnfd_info["cps"] = self.vnfd_info["cps"] * self.scale_inst_num
 132             for cp in self.vnfd_info["cps"]:
 133                 # TODO: how to set name for scale_out cp
 134                 cp["properties"]["name"] = cp["cp_id"] + str(uuid.uuid4())
 135                 cp_inst = PortInstModel.objects.filter(name__startswith=cp["cp_id"]).first()
 136                 if cp_inst:
 137                     cp["networkId"] = cp_inst.networkid
 138                     cp["subnetId"] = cp_inst.subnetworkid
 139                 else:
 140                     raise NFLCMException("CP(%s) does not exist" % cp["cp_id"])
 141             self.vnfd_info["vdus"] = self.vnfd_info["vdus"] * self.scale_inst_num
 142             for vdu in self.vnfd_info["vdus"]:
 143                 # TODO: how to set name for scale_out vdu
 144                 vdu["properties"]["name"] = vdu["properties"]["name"] + str(uuid.uuid4())
 145
 146             vim_cache = json.loads(self.vnf_insts[0].vimInfo)
 147             res_cache = json.loads(self.vnf_insts[0].resInfo)
 148             adaptor.create_vim_res(self.vnfd_info,
 149                                    self.do_notify_create_vim_res,
 150                                    vim_cache=vim_cache,
 151                                    res_cache=res_cache)
 152             self.vnf_insts.update(vimInfo=json.dumps(vim_cache),
 153                                   resInfo=json.dumps(res_cache))
 154         logger.debug("%s VNF resource finish", self.scale_type)
 155
 156     def send_notification(self):
 157         data = prepare_notification(nfinstid=self.nf_inst_id,
 158                                     jobid=self.job_id,
 159                                     operation=self.op_type,
 160                                     operation_state=OPERATION_STATE_TYPE.COMPLETED)
 161
 162         # TODO: need set changedExtConnectivity for data
 163         if self.is_scale_in:
 164             data["affectedVnfcs"] = self.affected_vnfcs
 165         else:
 166             for vm in self.scale_out_resource["vm"]:
 167                 self.set_affected_vnfcs(data["affectedVnfcs"], vm["res_id"])
 168
 169         logger.debug('Notify request data = %s' % data)
 170         NotificationsUtil().send_notification(data)
 171
 172     def rollback_operation(self):
 173         if self.is_scale_in:
 174             # SCALE_IN operaion does not support rollback
 175             return
 176         adaptor.delete_vim_res(self.scale_out_resource, self.do_notify_del_vim_res)
 177
 178     def set_location(self, grant_result):
 179         vim_connections = ignore_case_get(grant_result, "vimConnections")
 180         access_info = ignore_case_get(vim_connections[0], "accessInfo")
 181         tenant = ignore_case_get(access_info, "tenant")
 182         vimid = ignore_case_get(vim_connections[0], "vimId")
 183
 184         for resource_type in ['vdus', 'vls', 'cps', 'volume_storages']:
 185             for resource in ignore_case_get(self.vnfd_info, resource_type):
 186                 if "location_info" not in resource["properties"]:
 187                     resource["properties"]["location_info"] = {}
 188                 resource["properties"]["location_info"]["vimid"] = vimid
 189                 resource["properties"]["location_info"]["tenant"] = tenant
 190
 191     def do_notify_create_vim_res(self, res_type, ret):
 192         logger.debug('Scaling out [%s] resource' % res_type)
 193         resource_save_method = getattr(common, res_type + '_save')
 194         resource_save_method(self.job_id, self.nf_inst_id, ret)
 195         self.scale_out_resource[res_type].append(self.gen_del_resource(ret))
 196
 197     def do_notify_del_vim_res(self, res_type, res_id):
 198         logger.debug('Scaling in [%s] resource, resourceid [%s]', res_type, res_id)
 199         resource_type = RESOURCE_MAP.keys()[RESOURCE_MAP.values().index(res_type)]
 200         resource_table = globals().get(resource_type + 'InstModel')
 201         resource_table.objects.filter(instid=self.nf_inst_id, resourceid=res_id).delete()
 202         if res_type == "vm":
 203             VNFCInstModel.objects.filter(instid=self.nf_inst_id, vmid=res_id).delete()
 204
 205     def get_scale_in_resource(self, affected_vnfcs):
 206         scale_in_resource = {
 207             'volumn': [],
 208             'network': [],
 209             'subnet': [],
 210             'port': [],
 211             'flavor': [],
 212             'vm': []
 213         }
 214         scale_in_vms = VmInstModel.objects.filter(instid=self.nf_inst_id)
 215         vms_count = scale_in_vms.count()
 216         for index in range(self.scale_inst_num):
 217             vm_index = vms_count - index - 1
 218             scale_in_resource["vm"].append(self.gen_del_resource(scale_in_vms[vm_index]))
 219             self.set_affected_vnfcs(affected_vnfcs, scale_in_vms[vm_index].resourceid)
 220         return scale_in_resource
 221
 222     def gen_del_resource(self, res):
 223         is_dict = isinstance(res, dict)
 224         return {
 225             "vim_id": res["vimId"] if is_dict else res.vimid,
 226             "tenant_id": res["tenantId"] if is_dict else res.tenant,
 227             "res_id": res["id"] if is_dict else res.resourceid,
 228             "is_predefined": res["returnCode"] if is_dict else res.is_predefined
 229         }
 230
 231     def get_scale_step_delta(self):
 232         for policy in self.vnfd_info.get("policies", []):
 233             if policy.get("type") != "tosca.policies.nfv.ScalingAspects":
 234                 continue
 235             aspects = policy["properties"]["aspects"]
 236             if self.aspect_id in aspects:
 237                 return aspects.get(self.aspect_id).get("step_deltas")[0]
 238         raise NFLCMException("Aspect(%s) does not exist" % self.aspect_id)
 239
 240     def get_vdu_scale_aspect_deltas(self):
 241         for policy in self.vnfd_info.get("policies", []):
 242             if policy.get("type") != "tosca.policies.nfv.VduScalingAspectDeltas":
 243                 continue
 244             target = policy.get("targets")[0]
 245             deltas = policy["properties"]["deltas"]
 246             if self.step_delta in deltas:
 247                 num = int(deltas.get(self.step_delta).get("number_of_instances"))
 248                 return target, num
 249         raise NFLCMException("Aspect step delta(%s) does not exist" % self.step_delta)
 250
 251     def get_instance_range(self):
 252         for vdu in self.vnfd_info["vdus"]:
 253             if vdu["vdu_id"] == self.target_vdu:
 254                 vdu_profile = vdu["properties"]["vdu_profile"]
 255                 min_inst_num = int(vdu_profile["min_number_of_instances"])
 256                 max_inst_num = int(vdu_profile["max_number_of_instances"])
 257                 return min_inst_num, max_inst_num
 258         raise NFLCMException("VDU(%s) does not exist" % self.target_vdu)
 259
 260     def check_if_can_scale(self):
 261         cur_inst_num = VNFCInstModel.objects.filter(instid=self.nf_inst_id).count()
 262         if self.is_scale_in:
 263             if cur_inst_num - self.scale_inst_num < self.min_instance_num:
 264                 msg = "VNF(%s) cannot be scaled: less than min instance."
 265                 raise NFLCMException(msg % self.nf_inst_id)
 266         else:
 267             if cur_inst_num + self.scale_inst_num > self.max_instance_num:
 268                 msg = "VNF(%s) cannot be scaled: max instance exceeded."
 269                 raise NFLCMException(msg % self.nf_inst_id)
 270
 271     def set_affected_vnfcs(self, affected_vnfcs, vm_id):
 272         chgtype = CHANGE_TYPE.REMOVED if self.is_scale_in else CHANGE_TYPE.ADDED
 273         vnfcs = VNFCInstModel.objects.filter(instid=self.nf_inst_id, vmid=vm_id)
 274         vm = VmInstModel.objects.filter(instid=self.nf_inst_id, resourceid=vm_id)
 275         vm_resource = {}
 276         if vm:
 277             vm_resource = {
 278                 'vimConnectionId': vm[0].vimid,
 279                 'resourceId': vm[0].resourceid,
 280                 'vimLevelResourceType': 'vm'
 281             }
 282         if vnfcs:
 283             affected_vnfcs.append({
 284                 'id': vnfcs[0].vnfcinstanceid,
 285                 'vduId': vnfcs[0].vduid,
 286                 'changeType': chgtype,
 287                 'computeResource': vm_resource
 288             })
 289
 290     def vnf_scale_failed_handle(self, error_msg):
 291         logger.error('VNF scaling failed, detail message: %s', error_msg)
 292         self.vnf_insts.update(  # status=VNF_STATUS.FAILED,
 293             lastuptime=now_time())
 294         self.lcm_op_occ.notify_lcm(OPERATION_STATE_TYPE.FAILED, error_msg)
 295         JobUtil.add_job_status(self.job_id, 255, error_msg)