rwlaunchpad/plugins/rwnsm/rift/tasklets/rwnsmtasklet/rwnsmtasklet.py

   1 #
   2 #   Copyright 2016 RIFT.IO Inc
   3 #
   4 #   Licensed under the Apache License, Version 2.0 (the "License");
   5 #   you may not use this file except in compliance with the License.
   6 #   You may obtain a copy of the License at
   7 #
   8 #       http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #   Unless required by applicable law or agreed to in writing, software
  11 #   distributed under the License is distributed on an "AS IS" BASIS,
  12 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 #   See the License for the specific language governing permissions and
  14 #   limitations under the License.
  15 #
  16
  17 # vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
  18 import asyncio
  19 import ncclient
  20 import ncclient.asyncio_manager
  21 import os
  22 import shutil
  23 import sys
  24 import tempfile
  25 import time
  26 import uuid
  27 import yaml
  28 import requests
  29 import json
  30
  31
  32 from collections import deque
  33 from collections import defaultdict
  34 from enum import Enum
  35
  36 import gi
  37 gi.require_version('RwYang', '1.0')
  38 gi.require_version('RwNsdYang', '1.0')
  39 gi.require_version('RwDts', '1.0')
  40 gi.require_version('RwNsmYang', '1.0')
  41 gi.require_version('RwNsrYang', '1.0')
  42 gi.require_version('RwTypes', '1.0')
  43 gi.require_version('RwVlrYang', '1.0')
  44 gi.require_version('RwVnfrYang', '1.0')
  45 from gi.repository import (
  46     RwYang,
  47     RwNsrYang,
  48     NsrYang,
  49     NsdYang,
  50     RwVlrYang,
  51     VnfrYang,
  52     RwVnfrYang,
  53     RwNsmYang,
  54     RwsdnYang,
  55     RwDts as rwdts,
  56     RwTypes,
  57     ProtobufC,
  58 )
  59
  60 import rift.tasklets
  61 import rift.mano.ncclient
  62 import rift.mano.config_data.config
  63 import rift.mano.dts as mano_dts
  64
  65 from . import rwnsm_conman as conman
  66 from . import cloud
  67 from . import publisher
  68 from . import xpath
  69 from . import config_value_pool
  70 from . import rwvnffgmgr
  71 from . import scale_group
  72
  73
  74 class NetworkServiceRecordState(Enum):
  75     """ Network Service Record State """
  76     INIT = 101
  77     VL_INIT_PHASE = 102
  78     VNF_INIT_PHASE = 103
  79     VNFFG_INIT_PHASE = 104
  80     RUNNING = 106
  81     SCALING_OUT = 107
  82     SCALING_IN = 108
  83     TERMINATE = 109
  84     TERMINATE_RCVD = 110
  85     VL_TERMINATE_PHASE = 111
  86     VNF_TERMINATE_PHASE = 112
  87     VNFFG_TERMINATE_PHASE = 113
  88     TERMINATED = 114
  89     FAILED = 115
  90     VL_INSTANTIATE = 116
  91     VL_TERMINATE = 117
  92
  93
  94 class NetworkServiceRecordError(Exception):
  95     """ Network Service Record Error """
  96     pass
  97
  98
  99 class NetworkServiceDescriptorError(Exception):
 100     """ Network Service Descriptor Error """
 101     pass
 102
 103
 104 class VirtualNetworkFunctionRecordError(Exception):
 105     """ Virtual Network Function Record Error """
 106     pass
 107
 108
 109 class NetworkServiceDescriptorNotFound(Exception):
 110     """ Cannot find Network Service Descriptor"""
 111     pass
 112
 113
 114 class NetworkServiceDescriptorRefCountExists(Exception):
 115     """ Network Service Descriptor reference count exists """
 116     pass
 117
 118
 119 class NetworkServiceDescriptorUnrefError(Exception):
 120     """ Failed to unref a network service descriptor """
 121     pass
 122
 123
 124 class NsrInstantiationFailed(Exception):
 125     """ Failed to instantiate network service """
 126     pass
 127
 128
 129 class VnfInstantiationFailed(Exception):
 130     """ Failed to instantiate virtual network function"""
 131     pass
 132
 133
 134 class VnffgInstantiationFailed(Exception):
 135     """ Failed to instantiate virtual network function"""
 136     pass
 137
 138
 139 class VnfDescriptorError(Exception):
 140     """Failed to instantiate virtual network function"""
 141     pass
 142
 143
 144 class ScalingOperationError(Exception):
 145     pass
 146
 147
 148 class ScaleGroupMissingError(Exception):
 149     pass
 150
 151
 152 class PlacementGroupError(Exception):
 153     pass
 154
 155
 156 class NsrNsdUpdateError(Exception):
 157     pass
 158
 159
 160 class NsrVlUpdateError(NsrNsdUpdateError):
 161     pass
 162
 163
 164 class VlRecordState(Enum):
 165     """ VL Record State """
 166     INIT = 101
 167     INSTANTIATION_PENDING = 102
 168     ACTIVE = 103
 169     TERMINATE_PENDING = 104
 170     TERMINATED = 105
 171     FAILED = 106
 172
 173
 174 class VnffgRecordState(Enum):
 175     """ VNFFG Record State """
 176     INIT = 101
 177     INSTANTIATION_PENDING = 102
 178     ACTIVE = 103
 179     TERMINATE_PENDING = 104
 180     TERMINATED = 105
 181     FAILED = 106
 182
 183
 184 class VnffgRecord(object):
 185     """ Vnffg Records class"""
 186     SFF_DP_PORT = 4790
 187     SFF_MGMT_PORT = 5000
 188     def __init__(self, dts, log, loop, vnffgmgr, nsr, nsr_name, vnffgd_msg, sdn_account_name):
 189
 190         self._dts = dts
 191         self._log = log
 192         self._loop = loop
 193         self._vnffgmgr = vnffgmgr
 194         self._nsr = nsr
 195         self._nsr_name = nsr_name
 196         self._vnffgd_msg = vnffgd_msg
 197         if sdn_account_name is None:
 198             self._sdn_account_name = ''
 199         else:
 200             self._sdn_account_name = sdn_account_name
 201
 202         self._vnffgr_id = str(uuid.uuid4())
 203         self._vnffgr_rsp_id = list()
 204         self._vnffgr_state = VnffgRecordState.INIT
 205
 206     @property
 207     def id(self):
 208         """ VNFFGR id """
 209         return self._vnffgr_id
 210
 211     @property
 212     def state(self):
 213         """ state of this VNF """
 214         return self._vnffgr_state
 215
 216     def fetch_vnffgr(self):
 217         """
 218         Get VNFFGR message to be published
 219         """
 220
 221         if self._vnffgr_state == VnffgRecordState.INIT:
 222             vnffgr_dict = {"id": self._vnffgr_id,
 223                            "vnffgd_id_ref": self._vnffgd_msg.id,
 224                            "vnffgd_name_ref": self._vnffgd_msg.name,
 225                            "sdn_account": self._sdn_account_name,
 226                            "operational_status": 'init',
 227                            }
 228             vnffgr = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr.from_dict(vnffgr_dict)
 229         elif self._vnffgr_state == VnffgRecordState.TERMINATED:
 230             vnffgr_dict = {"id": self._vnffgr_id,
 231                            "vnffgd_id_ref": self._vnffgd_msg.id,
 232                            "vnffgd_name_ref": self._vnffgd_msg.name,
 233                            "sdn_account": self._sdn_account_name,
 234                            "operational_status": 'terminated',
 235                            }
 236             vnffgr = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr.from_dict(vnffgr_dict)
 237         else:
 238             try:
 239                 vnffgr = self._vnffgmgr.fetch_vnffgr(self._vnffgr_id)
 240             except Exception:
 241                 self._log.exception("Fetching VNFFGR for VNFFG with id %s failed", self._vnffgr_id)
 242                 self._vnffgr_state = VnffgRecordState.FAILED
 243                 vnffgr_dict = {"id": self._vnffgr_id,
 244                                "vnffgd_id_ref": self._vnffgd_msg.id,
 245                                "vnffgd_name_ref": self._vnffgd_msg.name,
 246                                "sdn_account": self._sdn_account_name,
 247                                "operational_status": 'failed',
 248                                }
 249                 vnffgr = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr.from_dict(vnffgr_dict)
 250
 251         return vnffgr
 252
 253     @asyncio.coroutine
 254     def vnffgr_create_msg(self):
 255         """ Virtual Link Record message for Creating VLR in VNS """
 256         vnffgr_dict = {"id": self._vnffgr_id,
 257                        "vnffgd_id_ref": self._vnffgd_msg.id,
 258                        "vnffgd_name_ref": self._vnffgd_msg.name,
 259                        "sdn_account": self._sdn_account_name,
 260                     }
 261         vnffgr = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr.from_dict(vnffgr_dict)
 262         for rsp in self._vnffgd_msg.rsp:
 263             vnffgr_rsp = vnffgr.rsp.add()
 264             vnffgr_rsp.id = str(uuid.uuid4())
 265             vnffgr_rsp.name = self._nsr.name + '.' + rsp.name
 266             self._vnffgr_rsp_id.append(vnffgr_rsp.id)
 267             vnffgr_rsp.vnffgd_rsp_id_ref =  rsp.id
 268             vnffgr_rsp.vnffgd_rsp_name_ref = rsp.name
 269             for rsp_cp_ref in rsp.vnfd_connection_point_ref:
 270                 vnfd =  [vnfr.vnfd for vnfr in self._nsr.vnfrs.values() if vnfr.vnfd.id == rsp_cp_ref.vnfd_id_ref]
 271                 self._log.debug("VNFD message during VNFFG instantiation is %s",vnfd)
 272                 if len(vnfd) > 0 and vnfd[0].has_field('service_function_type'):
 273                     self._log.debug("Service Function Type for VNFD ID %s is %s",rsp_cp_ref.vnfd_id_ref, vnfd[0].service_function_type)
 274                 else:
 275                     self._log.error("Service Function Type not available for VNFD ID %s; Skipping in chain",rsp_cp_ref.vnfd_id_ref)
 276                     continue
 277
 278                 vnfr_cp_ref =  vnffgr_rsp.vnfr_connection_point_ref.add()
 279                 vnfr_cp_ref.member_vnf_index_ref = rsp_cp_ref.member_vnf_index_ref
 280                 vnfr_cp_ref.hop_number = rsp_cp_ref.order
 281                 vnfr_cp_ref.vnfd_id_ref =rsp_cp_ref.vnfd_id_ref
 282                 vnfr_cp_ref.service_function_type = vnfd[0].service_function_type
 283                 for nsr_vnfr in self._nsr.vnfrs.values():
 284                    if (nsr_vnfr.vnfd.id == vnfr_cp_ref.vnfd_id_ref and
 285                       nsr_vnfr.member_vnf_index == vnfr_cp_ref.member_vnf_index_ref):
 286                        vnfr_cp_ref.vnfr_id_ref = nsr_vnfr.id
 287                        vnfr_cp_ref.vnfr_name_ref = nsr_vnfr.name
 288                        vnfr_cp_ref.vnfr_connection_point_ref = rsp_cp_ref.vnfd_connection_point_ref
 289
 290                        vnfr = yield from self._nsr.fetch_vnfr(nsr_vnfr.xpath)
 291                        self._log.debug(" Received VNFR is %s", vnfr)
 292                        while vnfr.operational_status != 'running':
 293                            self._log.info("Received vnf op status is %s; retrying",vnfr.operational_status)
 294                            if vnfr.operational_status == 'failed':
 295                                self._log.error("Fetching VNFR for  %s failed", vnfr.id)
 296                                raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self.id, vnfr.id))
 297                            yield from asyncio.sleep(2, loop=self._loop)
 298                            vnfr = yield from self._nsr.fetch_vnfr(nsr_vnfr.xpath)
 299                            self._log.debug("Received VNFR is %s", vnfr)
 300
 301                        vnfr_cp_ref.connection_point_params.mgmt_address =  vnfr.mgmt_interface.ip_address
 302                        for cp in vnfr.connection_point:
 303                            if cp.name == vnfr_cp_ref.vnfr_connection_point_ref:
 304                                vnfr_cp_ref.connection_point_params.port_id = cp.connection_point_id
 305                                vnfr_cp_ref.connection_point_params.name = self._nsr.name + '.' + cp.name
 306                                for vdu in vnfr.vdur:
 307                                    for ext_intf in vdu.external_interface:
 308                                        if ext_intf.name == vnfr_cp_ref.vnfr_connection_point_ref:
 309                                            vnfr_cp_ref.connection_point_params.vm_id =  vdu.vim_id
 310                                            self._log.debug("VIM ID for CP %s in VNFR %s is %s",cp.name,nsr_vnfr.id,
 311                                                             vnfr_cp_ref.connection_point_params.vm_id)
 312                                            break
 313
 314                                vnfr_cp_ref.connection_point_params.address =  cp.ip_address
 315                                vnfr_cp_ref.connection_point_params.port = VnffgRecord.SFF_DP_PORT
 316
 317         for vnffgd_classifier in self._vnffgd_msg.classifier:
 318             _rsp =  [rsp for rsp in vnffgr.rsp if rsp.vnffgd_rsp_id_ref == vnffgd_classifier.rsp_id_ref]
 319             if len(_rsp) > 0:
 320                 rsp_id_ref = _rsp[0].id
 321                 rsp_name = _rsp[0].name
 322             else:
 323                 self._log.error("RSP with ID %s not found during classifier creation for classifier id %s",vnffgd_classifier.rsp_id_ref,vnffgd_classifier.id)
 324                 continue
 325             vnffgr_classifier = vnffgr.classifier.add()
 326             vnffgr_classifier.id = vnffgd_classifier.id
 327             vnffgr_classifier.name =  self._nsr.name + '.' + vnffgd_classifier.name
 328             _rsp[0].classifier_name = vnffgr_classifier.name
 329             vnffgr_classifier.rsp_id_ref = rsp_id_ref
 330             vnffgr_classifier.rsp_name = rsp_name
 331             for nsr_vnfr in self._nsr.vnfrs.values():
 332                if (nsr_vnfr.vnfd.id == vnffgd_classifier.vnfd_id_ref and
 333                       nsr_vnfr.member_vnf_index == vnffgd_classifier.member_vnf_index_ref):
 334                        vnffgr_classifier.vnfr_id_ref = nsr_vnfr.id
 335                        vnffgr_classifier.vnfr_name_ref = nsr_vnfr.name
 336                        vnffgr_classifier.vnfr_connection_point_ref = vnffgd_classifier.vnfd_connection_point_ref
 337
 338                        if nsr_vnfr.vnfd.service_function_chain == 'CLASSIFIER':
 339                            vnffgr_classifier.sff_name = nsr_vnfr.name
 340
 341                        vnfr = yield from self._nsr.fetch_vnfr(nsr_vnfr.xpath)
 342                        self._log.debug(" Received VNFR is %s", vnfr)
 343                        while vnfr.operational_status != 'running':
 344                            self._log.info("Received vnf op status is %s; retrying",vnfr.operational_status)
 345                            if vnfr.operational_status == 'failed':
 346                                self._log.error("Fetching VNFR for  %s failed", vnfr.id)
 347                                raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self.id, vnfr.id))
 348                            yield from asyncio.sleep(2, loop=self._loop)
 349                            vnfr = yield from self._nsr.fetch_vnfr(nsr_vnfr.xpath)
 350                            self._log.debug("Received VNFR is %s", vnfr)
 351
 352                        for cp in vnfr.connection_point:
 353                            if cp.name == vnffgr_classifier.vnfr_connection_point_ref:
 354                                vnffgr_classifier.port_id = cp.connection_point_id
 355                                vnffgr_classifier.ip_address = cp.ip_address
 356                                for vdu in vnfr.vdur:
 357                                    for ext_intf in vdu.external_interface:
 358                                        if ext_intf.name == vnffgr_classifier.vnfr_connection_point_ref:
 359                                            vnffgr_classifier.vm_id =  vdu.vim_id
 360                                            self._log.debug("VIM ID for CP %s in VNFR %s is %s",cp.name,nsr_vnfr.id,
 361                                                             vnfr_cp_ref.connection_point_params.vm_id)
 362                                            break
 363
 364         self._log.info("VNFFGR msg to be sent is %s", vnffgr)
 365         return vnffgr
 366
 367     @asyncio.coroutine
 368     def vnffgr_nsr_sff_list(self):
 369         """ SFF List for VNFR """
 370         sff_list = {}
 371         sf_list = [nsr_vnfr.name for nsr_vnfr in self._nsr.vnfrs.values() if nsr_vnfr.vnfd.service_function_chain == 'SF']
 372
 373         for nsr_vnfr in self._nsr.vnfrs.values():
 374             if (nsr_vnfr.vnfd.service_function_chain == 'CLASSIFIER' or nsr_vnfr.vnfd.service_function_chain == 'SFF'):
 375                 vnfr = yield from self._nsr.fetch_vnfr(nsr_vnfr.xpath)
 376                 self._log.debug(" Received VNFR is %s", vnfr)
 377                 while vnfr.operational_status != 'running':
 378                     self._log.info("Received vnf op status is %s; retrying",vnfr.operational_status)
 379                     if vnfr.operational_status == 'failed':
 380                        self._log.error("Fetching VNFR for  %s failed", vnfr.id)
 381                        raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self.id, vnfr.id))
 382                     yield from asyncio.sleep(2, loop=self._loop)
 383                     vnfr = yield from self._nsr.fetch_vnfr(nsr_vnfr.xpath)
 384                     self._log.debug("Received VNFR is %s", vnfr)
 385
 386                 sff =  RwsdnYang.VNFFGSff()
 387                 sff_list[nsr_vnfr.vnfd.id] = sff
 388                 sff.name = nsr_vnfr.name
 389                 sff.function_type = nsr_vnfr.vnfd.service_function_chain
 390
 391                 sff.mgmt_address = vnfr.mgmt_interface.ip_address
 392                 sff.mgmt_port = VnffgRecord.SFF_MGMT_PORT
 393                 for cp in vnfr.connection_point:
 394                     sff_dp = sff.dp_endpoints.add()
 395                     sff_dp.name = self._nsr.name + '.' + cp.name
 396                     sff_dp.address = cp.ip_address
 397                     sff_dp.port  = VnffgRecord.SFF_DP_PORT
 398                 if nsr_vnfr.vnfd.service_function_chain == 'SFF':
 399                     for sf_name in sf_list:
 400                         _sf = sff.vnfr_list.add()
 401                         _sf.vnfr_name = sf_name
 402
 403         return sff_list
 404
 405     @asyncio.coroutine
 406     def instantiate(self):
 407         """ Instantiate this VNFFG """
 408
 409         self._log.info("Instaniating VNFFGR with vnffgd %s",
 410                        self._vnffgd_msg)
 411
 412
 413         vnffgr_request = yield from self.vnffgr_create_msg()
 414         vnffg_sff_list = yield from self.vnffgr_nsr_sff_list()
 415
 416         try:
 417             vnffgr = self._vnffgmgr.create_vnffgr(vnffgr_request,self._vnffgd_msg.classifier,vnffg_sff_list)
 418         except Exception as e:
 419             self._log.exception("VNFFG instantiation failed: %s", str(e))
 420             self._vnffgr_state = VnffgRecordState.FAILED
 421             raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFFGR %s failure" % (self.id, vnffgr_request.id))
 422
 423         self._vnffgr_state = VnffgRecordState.INSTANTIATION_PENDING
 424
 425         self._log.info("Instantiated VNFFGR :%s", vnffgr)
 426         self._vnffgr_state = VnffgRecordState.ACTIVE
 427
 428         self._log.info("Invoking update_state to update NSR state for NSR ID: %s", self._nsr.id)
 429         yield from self._nsr.update_state()
 430
 431     def vnffgr_in_vnffgrm(self):
 432         """ Is there a VNFR record in VNFM """
 433         if (self._vnffgr_state == VnffgRecordState.ACTIVE or
 434                 self._vnffgr_state == VnffgRecordState.INSTANTIATION_PENDING or
 435                 self._vnffgr_state == VnffgRecordState.FAILED):
 436             return True
 437
 438         return False
 439
 440     @asyncio.coroutine
 441     def terminate(self):
 442         """ Terminate this VNFFGR """
 443         if not self.vnffgr_in_vnffgrm():
 444             self._log.error("Ignoring terminate request for id %s in state %s",
 445                             self.id, self._vnffgr_state)
 446             return
 447
 448         self._log.info("Terminating VNFFGR id:%s", self.id)
 449         self._vnffgr_state = VnffgRecordState.TERMINATE_PENDING
 450
 451         self._vnffgmgr.terminate_vnffgr(self._vnffgr_id)
 452
 453         self._vnffgr_state = VnffgRecordState.TERMINATED
 454         self._log.debug("Terminated VNFFGR id:%s", self.id)
 455
 456
 457 class VirtualLinkRecord(object):
 458     """ Virtual Link Records class"""
 459     XPATH = "D,/vlr:vlr-catalog/vlr:vlr"
 460     @staticmethod
 461     @asyncio.coroutine
 462     def create_record(dts, log, loop, nsr_name, vld_msg, cloud_account_name, om_datacenter, ip_profile, nsr_id, restart_mode=False):
 463         """Creates a new VLR object based on the given data.
 464
 465         If restart mode is enabled, then we look for existing records in the
 466         DTS and create a VLR records using the exiting data(ID)
 467
 468         Returns:
 469             VirtualLinkRecord
 470         """
 471         vlr_obj = VirtualLinkRecord(
 472                       dts,
 473                       log,
 474                       loop,
 475                       nsr_name,
 476                       vld_msg,
 477                       cloud_account_name,
 478                       om_datacenter,
 479                       ip_profile,
 480                       nsr_id,
 481                       )
 482
 483         if restart_mode:
 484             res_iter = yield from dts.query_read(
 485                               "D,/vlr:vlr-catalog/vlr:vlr",
 486                               rwdts.XactFlag.MERGE)
 487
 488             for fut in res_iter:
 489                 response = yield from fut
 490                 vlr = response.result
 491
 492                 # Check if the record is already present, if so use the ID of
 493                 # the existing record. Since the name of the record is uniquely
 494                 # formed we can use it as a search key!
 495                 if vlr.name == vlr_obj.name:
 496                     vlr_obj.reset_id(vlr.id)
 497                     break
 498
 499         return vlr_obj
 500
 501     def __init__(self, dts, log, loop, nsr_name, vld_msg, cloud_account_name, om_datacenter, ip_profile, nsr_id):
 502         self._dts = dts
 503         self._log = log
 504         self._loop = loop
 505         self._nsr_name = nsr_name
 506         self._vld_msg = vld_msg
 507         self._cloud_account_name = cloud_account_name
 508         self._om_datacenter_name = om_datacenter
 509         self._assigned_subnet = None
 510         self._nsr_id = nsr_id
 511         self._ip_profile = ip_profile
 512         self._vlr_id = str(uuid.uuid4())
 513         self._state = VlRecordState.INIT
 514         self._prev_state = None
 515         self._create_time = int(time.time())
 516
 517     @property
 518     def xpath(self):
 519         """ path for this object """
 520         return "D,/vlr:vlr-catalog/vlr:vlr[vlr:id = '{}']".format(self._vlr_id)
 521
 522     @property
 523     def id(self):
 524         """ VLR id """
 525         return self._vlr_id
 526
 527     @property
 528     def nsr_name(self):
 529         """ Get NSR name for this VL """
 530         return self.nsr_name
 531
 532     @property
 533     def vld_msg(self):
 534         """ Virtual Link Desciptor """
 535         return self._vld_msg
 536
 537     @property
 538     def assigned_subnet(self):
 539         """ Subnet assigned to this VL"""
 540         return self._assigned_subnet
 541
 542     @property
 543     def name(self):
 544         """
 545         Get the name for this VLR.
 546         VLR name is "nsr name:VLD name"
 547         """
 548         if self.vld_msg.vim_network_name:
 549             return self.vld_msg.vim_network_name
 550         elif self.vld_msg.name == "multisite":
 551             # This is a temporary hack to identify manually provisioned inter-site network
 552             return self.vld_msg.name
 553         else:
 554             return self._nsr_name + "." + self.vld_msg.name
 555
 556     @property
 557     def cloud_account_name(self):
 558         """ Cloud account that this VLR should be created in """
 559         return self._cloud_account_name
 560
 561     @property
 562     def om_datacenter_name(self):
 563         """ Datacenter  that this VLR should be created in """
 564         return self._om_datacenter_name
 565
 566     @staticmethod
 567     def vlr_xpath(vlr):
 568         """ Get the VLR path from VLR """
 569         return (VirtualLinkRecord.XPATH + "[vlr:id = '{}']").format(vlr.id)
 570
 571     @property
 572     def state(self):
 573         """ VLR state """
 574         return self._state
 575
 576     @state.setter
 577     def state(self, value):
 578         """ VLR set state """
 579         self._state = value
 580
 581     @property
 582     def prev_state(self):
 583         """ VLR previous state """
 584         return self._prev_state
 585
 586     @prev_state.setter
 587     def prev_state(self, value):
 588         """ VLR set previous state """
 589         self._prev_state = value
 590
 591     @property
 592     def vlr_msg(self):
 593         """ Virtual Link Record message for Creating VLR in VNS """
 594         vld_fields = ["short_name",
 595                       "vendor",
 596                       "description",
 597                       "version",
 598                       "type_yang",
 599                       "vim_network_name",
 600                       "provider_network"]
 601
 602         vld_copy_dict = {k: v for k, v in self.vld_msg.as_dict().items()
 603                          if k in vld_fields}
 604
 605         vlr_dict = {"id": self._vlr_id,
 606                     "nsr_id_ref": self._nsr_id,
 607                     "vld_ref": self.vld_msg.id,
 608                     "name": self.name,
 609                     "create_time": self._create_time,
 610                     "cloud_account": self.cloud_account_name,
 611                     "om_datacenter": self.om_datacenter_name,
 612                     }
 613
 614         if self._ip_profile and self._ip_profile.has_field('ip_profile_params'):
 615             vlr_dict['ip_profile_params' ] = self._ip_profile.ip_profile_params.as_dict()
 616
 617         vlr_dict.update(vld_copy_dict)
 618         vlr = RwVlrYang.YangData_Vlr_VlrCatalog_Vlr.from_dict(vlr_dict)
 619         return vlr
 620
 621     def reset_id(self, vlr_id):
 622         self._vlr_id = vlr_id
 623
 624     def create_nsr_vlr_msg(self, vnfrs):
 625         """ The VLR message"""
 626         nsr_vlr = RwNsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_Vlr()
 627         nsr_vlr.vlr_ref = self._vlr_id
 628         nsr_vlr.assigned_subnet = self.assigned_subnet
 629         nsr_vlr.cloud_account = self.cloud_account_name
 630         nsr_vlr.om_datacenter = self.om_datacenter_name
 631
 632         for conn in self.vld_msg.vnfd_connection_point_ref:
 633             for vnfr in vnfrs:
 634                 if (vnfr.vnfd.id == conn.vnfd_id_ref and
 635                         vnfr.member_vnf_index == conn.member_vnf_index_ref and
 636                         self.cloud_account_name == vnfr.cloud_account_name and
 637                         self.om_datacenter_name == vnfr.om_datacenter_name):
 638                     cp_entry = nsr_vlr.vnfr_connection_point_ref.add()
 639                     cp_entry.vnfr_id = vnfr.id
 640                     cp_entry.connection_point = conn.vnfd_connection_point_ref
 641
 642         return nsr_vlr
 643
 644     @asyncio.coroutine
 645     def instantiate(self):
 646         """ Instantiate this VL """
 647         self._log.debug("Instaniating VLR key %s, vld %s",
 648                         self.xpath, self._vld_msg)
 649         vlr = None
 650         self._state = VlRecordState.INSTANTIATION_PENDING
 651         self._log.debug("Executing VL create path:%s msg:%s",
 652                         self.xpath, self.vlr_msg)
 653
 654         with self._dts.transaction(flags=0) as xact:
 655             block = xact.block_create()
 656             block.add_query_create(self.xpath, self.vlr_msg)
 657             self._log.debug("Executing VL create path:%s msg:%s",
 658                             self.xpath, self.vlr_msg)
 659             res_iter = yield from block.execute(now=True)
 660             for ent in res_iter:
 661                 res = yield from ent
 662                 vlr = res.result
 663
 664             if vlr is None:
 665                 self._state = VlRecordState.FAILED
 666                 raise NsrInstantiationFailed("Failed NS %s instantiation due to empty response" % self.id)
 667
 668         if vlr.operational_status == 'failed':
 669             self._log.debug("NS Id:%s VL creation failed for vlr id %s", self.id, vlr.id)
 670             self._state = VlRecordState.FAILED
 671             raise NsrInstantiationFailed("Failed VL %s instantiation (%s)" % (vlr.id, vlr.operational_status_details))
 672
 673         self._log.info("Instantiated VL with xpath %s and vlr:%s",
 674                        self.xpath, vlr)
 675         self._state = VlRecordState.ACTIVE
 676         self._assigned_subnet = vlr.assigned_subnet
 677
 678     def vlr_in_vns(self):
 679         """ Is there a VLR record in VNS """
 680         if (self._state == VlRecordState.ACTIVE or
 681             self._state == VlRecordState.INSTANTIATION_PENDING or
 682             self._state == VlRecordState.TERMINATE_PENDING or
 683             self._state == VlRecordState.FAILED):
 684             return True
 685
 686         return False
 687
 688     @asyncio.coroutine
 689     def terminate(self):
 690         """ Terminate this VL """
 691         if not self.vlr_in_vns():
 692             self._log.debug("Ignoring terminate request for id %s in state %s",
 693                             self.id, self._state)
 694             return
 695
 696         self._log.debug("Terminating VL id:%s", self.id)
 697         self._state = VlRecordState.TERMINATE_PENDING
 698
 699         with self._dts.transaction(flags=0) as xact:
 700             block = xact.block_create()
 701             block.add_query_delete(self.xpath)
 702             yield from block.execute(flags=0, now=True)
 703
 704         self._state = VlRecordState.TERMINATED
 705         self._log.debug("Terminated VL id:%s", self.id)
 706
 707
 708 class VnfRecordState(Enum):
 709     """ Vnf Record State """
 710     INIT = 101
 711     INSTANTIATION_PENDING = 102
 712     ACTIVE = 103
 713     TERMINATE_PENDING = 104
 714     TERMINATED = 105
 715     FAILED = 106
 716
 717
 718 class VirtualNetworkFunctionRecord(object):
 719     """ Virtual Network Function Record class"""
 720     XPATH = "D,/vnfr:vnfr-catalog/vnfr:vnfr"
 721
 722     @staticmethod
 723     @asyncio.coroutine
 724     def create_record(dts, log, loop, vnfd, const_vnfd_msg, nsd_id, nsr_name,
 725                 cloud_account_name, om_datacenter_name, nsr_id, group_name, group_instance_id,
 726                 placement_groups, restart_mode=False):
 727         """Creates a new VNFR object based on the given data.
 728
 729         If restart mode is enabled, then we look for existing records in the
 730         DTS and create a VNFR records using the exiting data(ID)
 731
 732         Returns:
 733             VirtualNetworkFunctionRecord
 734         """
 735         vnfr_obj = VirtualNetworkFunctionRecord(
 736                           dts,
 737                           log,
 738                           loop,
 739                           vnfd,
 740                           const_vnfd_msg,
 741                           nsd_id,
 742                           nsr_name,
 743                           cloud_account_name,
 744                           om_datacenter_name,
 745                           nsr_id,
 746                           group_name,
 747                           group_instance_id,
 748                           placement_groups,
 749                           restart_mode=restart_mode)
 750
 751         if restart_mode:
 752             res_iter = yield from dts.query_read(
 753                               "D,/vnfr:vnfr-catalog/vnfr:vnfr",
 754                               rwdts.XactFlag.MERGE)
 755
 756             for fut in res_iter:
 757                 response = yield from fut
 758                 vnfr = response.result
 759
 760                 if vnfr.name == vnfr_obj.name:
 761                     vnfr_obj.reset_id(vnfr.id)
 762                     break
 763
 764         return vnfr_obj
 765
 766     def __init__(self,
 767                  dts,
 768                  log,
 769                  loop,
 770                  vnfd,
 771                  const_vnfd_msg,
 772                  nsd_id,
 773                  nsr_name,
 774                  cloud_account_name,
 775                  om_datacenter_name,
 776                  nsr_id,
 777                  group_name=None,
 778                  group_instance_id=None,
 779                  placement_groups = [],
 780                  restart_mode = False):
 781         self._dts = dts
 782         self._log = log
 783         self._loop = loop
 784         self._vnfd = vnfd
 785         self._const_vnfd_msg = const_vnfd_msg
 786         self._nsd_id = nsd_id
 787         self._nsr_name = nsr_name
 788         self._nsr_id = nsr_id
 789         self._cloud_account_name = cloud_account_name
 790         self._om_datacenter_name = om_datacenter_name
 791         self._group_name = group_name
 792         self._group_instance_id = group_instance_id
 793         self._placement_groups = placement_groups
 794         self._config_status = NsrYang.ConfigStates.INIT
 795         self._create_time = int(time.time())
 796
 797         self._prev_state = VnfRecordState.INIT
 798         self._state = VnfRecordState.INIT
 799         self._state_failed_reason = None
 800
 801         self.config_store = rift.mano.config_data.config.ConfigStore(self._log)
 802         self.configure()
 803
 804         self._vnfr_id = str(uuid.uuid4())
 805         self._name = None
 806         self._vnfr_msg = self.create_vnfr_msg()
 807         self._log.debug("Set VNFR {} config type to {}".
 808                         format(self.name, self.config_type))
 809         self.restart_mode = restart_mode
 810
 811
 812         if group_name is None and group_instance_id is not None:
 813             raise ValueError("Group instance id must not be provided with an empty group name")
 814
 815     @property
 816     def id(self):
 817         """ VNFR id """
 818         return self._vnfr_id
 819
 820     @property
 821     def xpath(self):
 822         """ VNFR xpath """
 823         return "D,/vnfr:vnfr-catalog/vnfr:vnfr[vnfr:id = '{}']".format(self.id)
 824
 825     @property
 826     def vnfr_msg(self):
 827         """ VNFR message """
 828         return self._vnfr_msg
 829
 830     @property
 831     def const_vnfr_msg(self):
 832         """ VNFR message """
 833         return RwNsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_ConstituentVnfrRef(vnfr_id=self.id,cloud_account=self.cloud_account_name,om_datacenter=self._om_datacenter_name)
 834
 835     @property
 836     def vnfd(self):
 837         """ vnfd """
 838         return self._vnfd
 839
 840     @property
 841     def cloud_account_name(self):
 842         """ Cloud account that this VNF should be created in """
 843         return self._cloud_account_name
 844
 845     @property
 846     def om_datacenter_name(self):
 847         """ Datacenter that this VNF should be created in """
 848         return self._om_datacenter_name
 849
 850
 851     @property
 852     def active(self):
 853         """ Is this VNF actve """
 854         return True if self._state == VnfRecordState.ACTIVE else False
 855
 856     @property
 857     def state(self):
 858         """ state of this VNF """
 859         return self._state
 860
 861     @property
 862     def state_failed_reason(self):
 863         """ Error message in case this VNF is in failed state """
 864         return self._state_failed_reason
 865
 866     @property
 867     def member_vnf_index(self):
 868         """ Member VNF index """
 869         return self._const_vnfd_msg.member_vnf_index
 870
 871     @property
 872     def nsr_name(self):
 873         """ NSR name"""
 874         return self._nsr_name
 875
 876     @property
 877     def name(self):
 878         """ Name of this VNFR """
 879         if self._name is not None:
 880             return self._name
 881
 882         name_tags = [self._nsr_name]
 883
 884         if self._group_name is not None:
 885             name_tags.append(self._group_name)
 886
 887         if self._group_instance_id is not None:
 888             name_tags.append(str(self._group_instance_id))
 889
 890         name_tags.extend([self.vnfd.name, str(self.member_vnf_index)])
 891
 892         self._name = "__".join(name_tags)
 893
 894         return self._name
 895
 896     @staticmethod
 897     def vnfr_xpath(vnfr):
 898         """ Get the VNFR path from VNFR """
 899         return (VirtualNetworkFunctionRecord.XPATH + "[vnfr:id = '{}']").format(vnfr.id)
 900
 901     @property
 902     def config_type(self):
 903         cfg_types = ['netconf', 'juju', 'script']
 904         for method in cfg_types:
 905             if self._vnfd.vnf_configuration.has_field(method):
 906                 return method
 907         return 'none'
 908
 909     @property
 910     def config_status(self):
 911         """Return the config status as YANG ENUM string"""
 912         self._log.debug("Map VNFR {} config status {} ({})".
 913                         format(self.name, self._config_status, self.config_type))
 914         if self.config_type == 'none':
 915             return 'config_not_needed'
 916         elif self._config_status == NsrYang.ConfigStates.CONFIGURED:
 917             return 'configured'
 918         elif self._config_status == NsrYang.ConfigStates.FAILED:
 919             return 'failed'
 920
 921         return 'configuring'
 922
 923     def set_state(self, state):
 924         """ set the state of this object """
 925         self._prev_state = self._state
 926         self._state = state
 927
 928     def reset_id(self, vnfr_id):
 929         self._vnfr_id = vnfr_id
 930         self._vnfr_msg = self.create_vnfr_msg()
 931
 932     def configure(self):
 933         self.config_store.merge_vnfd_config(
 934                     self._nsd_id,
 935                     self._vnfd,
 936                     self.member_vnf_index,
 937                     )
 938
 939     def create_vnfr_msg(self):
 940         """ VNFR message for this VNFR """
 941         vnfd_fields = [
 942                 "short_name",
 943                 "vendor",
 944                 "description",
 945                 "version",
 946                 "type_yang",
 947                 ]
 948         vnfd_copy_dict = {k: v for k, v in self._vnfd.as_dict().items() if k in vnfd_fields}
 949         vnfr_dict = {
 950                 "id": self.id,
 951                 "nsr_id_ref": self._nsr_id,
 952                 "name": self.name,
 953                 "cloud_account": self._cloud_account_name,
 954                 "om_datacenter": self._om_datacenter_name,
 955                 "config_status": self.config_status
 956                 }
 957         vnfr_dict.update(vnfd_copy_dict)
 958
 959         vnfr = RwVnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr.from_dict(vnfr_dict)
 960
 961         vnfr.vnfd = VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_Vnfd.from_dict(self.vnfd.as_dict(),
 962                                                                           ignore_missing_keys=True)
 963         vnfr.member_vnf_index_ref = self.member_vnf_index
 964         vnfr.vnf_configuration.from_dict(self._vnfd.vnf_configuration.as_dict())
 965
 966         if self._vnfd.mgmt_interface.has_field("port"):
 967             vnfr.mgmt_interface.port = self._vnfd.mgmt_interface.port
 968
 969         for group_info in self._placement_groups:
 970             group = vnfr.placement_groups_info.add()
 971             group.from_dict(group_info.as_dict())
 972
 973         # UI expects the monitoring param field to exist
 974         vnfr.monitoring_param = []
 975
 976         self._log.debug("Get vnfr_msg for VNFR {} : {}".format(self.name, vnfr))
 977         return vnfr
 978
 979     @asyncio.coroutine
 980     def update_vnfm(self):
 981         self._log.debug("Send an update to VNFM for VNFR {} with {}".
 982                         format(self.name, self.vnfr_msg))
 983         yield from self._dts.query_update(
 984                 self.xpath,
 985                 rwdts.XactFlag.TRACE,
 986                 self.vnfr_msg
 987                 )
 988
 989     def get_config_status(self):
 990         """Return the config status as YANG ENUM"""
 991         return self._config_status
 992
 993     @asyncio.coroutine
 994     def set_config_status(self, status):
 995
 996         def status_to_string(status):
 997             status_dc = {
 998                 NsrYang.ConfigStates.INIT : 'init',
 999                 NsrYang.ConfigStates.CONFIGURING : 'configuring',
1000                 NsrYang.ConfigStates.CONFIG_NOT_NEEDED : 'config_not_needed',
1001                 NsrYang.ConfigStates.CONFIGURED : 'configured',
1002                 NsrYang.ConfigStates.FAILED : 'failed',
1003             }
1004
1005             return status_dc[status]
1006
1007         self._log.debug("Update VNFR {} from {} ({}) to {}".
1008                         format(self.name, self._config_status,
1009                                self.config_type, status))
1010         if self._config_status == NsrYang.ConfigStates.CONFIGURED:
1011             self._log.error("Updating already configured VNFR {}".
1012                             format(self.name))
1013             return
1014
1015         if self._config_status != status:
1016             try:
1017                 self._config_status = status
1018                 # I don't think this is used. Original implementor can check.
1019                 # Caused Exception, so corrected it by status_to_string
1020                 # But not sure whats the use of this variable?
1021                 self.vnfr_msg.config_status = status_to_string(status)
1022             except Exception as e:
1023                 self._log.error("Exception=%s", str(e))
1024                 pass
1025
1026             self._log.debug("Updated VNFR {} status to {}".format(self.name, status))
1027
1028             if self._config_status != NsrYang.ConfigStates.INIT:
1029                 try:
1030                     # Publish only after VNFM has the VNFR created
1031                     yield from self.update_vnfm()
1032                 except Exception as e:
1033                     self._log.error("Exception updating VNFM with new status {} of VNFR {}: {}".
1034                                 format(status, self.name, e))
1035                     self._log.exception(e)
1036
1037     def is_configured(self):
1038         if self.config_type == 'none':
1039             return True
1040
1041         if self._config_status == NsrYang.ConfigStates.CONFIGURED:
1042             return True
1043
1044         return False
1045
1046     @asyncio.coroutine
1047     def instantiate(self, nsr):
1048         """ Instantiate this VNFR"""
1049
1050         self._log.debug("Instaniating VNFR key %s, vnfd %s",
1051                         self.xpath, self._vnfd)
1052
1053         self._log.debug("Create VNF with xpath %s and vnfr %s",
1054                         self.xpath, self.vnfr_msg)
1055
1056         self.set_state(VnfRecordState.INSTANTIATION_PENDING)
1057
1058         def find_vlr_for_cp(conn):
1059             """ Find VLR for the given connection point """
1060             for vlr in nsr.vlrs:
1061                 for vnfd_cp in vlr.vld_msg.vnfd_connection_point_ref:
1062                     if (vnfd_cp.vnfd_id_ref == self._vnfd.id and
1063                             vnfd_cp.vnfd_connection_point_ref == conn.name and
1064                             vnfd_cp.member_vnf_index_ref == self.member_vnf_index and
1065                              vlr.cloud_account_name == self.cloud_account_name):
1066                         self._log.debug("Found VLR for cp_name:%s and vnf-index:%d",
1067                                         conn.name, self.member_vnf_index)
1068                         return vlr
1069             return None
1070
1071         # For every connection point in the VNFD fill in the identifier
1072         for conn_p in self._vnfd.connection_point:
1073             cpr = VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_ConnectionPoint()
1074             cpr.name = conn_p.name
1075             cpr.type_yang = conn_p.type_yang
1076             cpr.port_security_enabled = conn_p.port_security_enabled
1077             vlr_ref = find_vlr_for_cp(conn_p)
1078             if vlr_ref is None:
1079                 msg = "Failed to find VLR for cp = %s" % conn_p.name
1080                 self._log.debug("%s", msg)
1081 #                raise VirtualNetworkFunctionRecordError(msg)
1082                 continue
1083
1084             cpr.vlr_ref = vlr_ref.id
1085             self.vnfr_msg.connection_point.append(cpr)
1086             self._log.debug("Connection point [%s] added, vnf id=%s vnfd id=%s",
1087                             cpr, self.vnfr_msg.id, self.vnfr_msg.vnfd.id)
1088
1089         if not self.restart_mode:
1090             yield from self._dts.query_create(self.xpath,
1091                                               0,   # this is sub
1092                                               self.vnfr_msg)
1093         else:
1094             yield from self._dts.query_update(self.xpath,
1095                                               0,
1096                                               self.vnfr_msg)
1097
1098         self._log.info("Created VNF with xpath %s and vnfr %s",
1099                        self.xpath, self.vnfr_msg)
1100
1101         self._log.info("Instantiated VNFR with xpath %s and vnfd %s, vnfr %s",
1102                        self.xpath, self._vnfd, self.vnfr_msg)
1103
1104     @asyncio.coroutine
1105     def update_state(self, vnfr_msg):
1106         """ Update this VNFR"""
1107         if vnfr_msg.operational_status == "running":
1108             if self.vnfr_msg.operational_status != "running":
1109                 yield from self.is_active()
1110         elif vnfr_msg.operational_status == "failed":
1111             yield from self.instantiation_failed(failed_reason=vnfr_msg.operational_status_details)
1112
1113     @asyncio.coroutine
1114     def is_active(self):
1115         """ This VNFR is active """
1116         self._log.debug("VNFR %s is active", self._vnfr_id)
1117         self.set_state(VnfRecordState.ACTIVE)
1118
1119     @asyncio.coroutine
1120     def instantiation_failed(self, failed_reason=None):
1121         """ This VNFR instantiation failed"""
1122         self._log.error("VNFR %s instantiation failed", self._vnfr_id)
1123         self.set_state(VnfRecordState.FAILED)
1124         self._state_failed_reason = failed_reason
1125
1126     def vnfr_in_vnfm(self):
1127         """ Is there a VNFR record in VNFM """
1128         if (self._state == VnfRecordState.ACTIVE or
1129                 self._state == VnfRecordState.INSTANTIATION_PENDING or
1130                 self._state == VnfRecordState.FAILED):
1131             return True
1132
1133         return False
1134
1135     @asyncio.coroutine
1136     def terminate(self):
1137         """ Terminate this VNF """
1138         if not self.vnfr_in_vnfm():
1139             self._log.debug("Ignoring terminate request for id %s in state %s",
1140                             self.id, self._state)
1141             return
1142
1143         self._log.debug("Terminating VNF id:%s", self.id)
1144         self.set_state(VnfRecordState.TERMINATE_PENDING)
1145         with self._dts.transaction(flags=0) as xact:
1146             block = xact.block_create()
1147             block.add_query_delete(self.xpath)
1148             yield from block.execute(flags=0)
1149         self.set_state(VnfRecordState.TERMINATED)
1150         self._log.debug("Terminated VNF id:%s", self.id)
1151
1152
1153 class NetworkServiceStatus(object):
1154     """ A class representing the Network service's status """
1155     MAX_EVENTS_RECORDED = 10
1156     """ Network service Status class"""
1157     def __init__(self, dts, log, loop):
1158         self._dts = dts
1159         self._log = log
1160         self._loop = loop
1161
1162         self._state = NetworkServiceRecordState.INIT
1163         self._events = deque([])
1164
1165     @asyncio.coroutine
1166     def create_notification(self, evt, evt_desc, evt_details):
1167         xp = "N,/rw-nsr:nsm-notification"
1168         notif = RwNsrYang.YangNotif_RwNsr_NsmNotification()
1169         notif.event = evt
1170         notif.description = evt_desc
1171         notif.details = evt_details if evt_details is not None else None
1172
1173         yield from self._dts.query_create(xp, rwdts.XactFlag.ADVISE, notif)
1174         self._log.info("Notification called by creating dts query: %s", notif)
1175
1176     def record_event(self, evt, evt_desc, evt_details):
1177         """ Record an event """
1178         self._log.debug("Recording event - evt %s, evt_descr %s len = %s",
1179                         evt, evt_desc, len(self._events))
1180         if len(self._events) >= NetworkServiceStatus.MAX_EVENTS_RECORDED:
1181             self._events.popleft()
1182         self._events.append((int(time.time()), evt, evt_desc,
1183                              evt_details if evt_details is not None else None))
1184
1185         self._loop.create_task(self.create_notification(evt,evt_desc,evt_details))
1186
1187     def set_state(self, state):
1188         """ set the state of this status object """
1189         self._state = state
1190
1191     def yang_str(self):
1192         """ Return the state as a yang enum string """
1193         state_to_str_map = {"INIT": "init",
1194                             "VL_INIT_PHASE": "vl_init_phase",
1195                             "VNF_INIT_PHASE": "vnf_init_phase",
1196                             "VNFFG_INIT_PHASE": "vnffg_init_phase",
1197                             "SCALING_GROUP_INIT_PHASE": "scaling_group_init_phase",
1198                             "RUNNING": "running",
1199                             "SCALING_OUT": "scaling_out",
1200                             "SCALING_IN": "scaling_in",
1201                             "TERMINATE_RCVD": "terminate_rcvd",
1202                             "TERMINATE": "terminate",
1203                             "VL_TERMINATE_PHASE": "vl_terminate_phase",
1204                             "VNF_TERMINATE_PHASE": "vnf_terminate_phase",
1205                             "VNFFG_TERMINATE_PHASE": "vnffg_terminate_phase",
1206                             "TERMINATED": "terminated",
1207                             "FAILED": "failed",
1208                             "VL_INSTANTIATE": "vl_instantiate",
1209                             "VL_TERMINATE": "vl_terminate",
1210         }
1211         return state_to_str_map[self._state.name]
1212
1213     @property
1214     def state(self):
1215         """ State of this status object """
1216         return self._state
1217
1218     @property
1219     def msg(self):
1220         """ Network Service Record as a message"""
1221         event_list = []
1222         idx = 1
1223         for entry in self._events:
1224             event = RwNsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_OperationalEvents()
1225             event.id = idx
1226             idx += 1
1227             event.timestamp, event.event, event.description, event.details = entry
1228             event_list.append(event)
1229         return event_list
1230
1231
1232 class NetworkServiceRecord(object):
1233     """ Network service record """
1234     XPATH = "D,/nsr:ns-instance-opdata/nsr:nsr"
1235
1236     def __init__(self, dts, log, loop, nsm, nsm_plugin, nsr_cfg_msg, sdn_account_name, key_pairs, restart_mode=False,
1237                  vlr_handler=None):
1238         self._dts = dts
1239         self._log = log
1240         self._loop = loop
1241         self._nsm = nsm
1242         self._nsr_cfg_msg = nsr_cfg_msg
1243         self._nsm_plugin = nsm_plugin
1244         self._sdn_account_name = sdn_account_name
1245         self._vlr_handler = vlr_handler
1246
1247         self._nsd = None
1248         self._nsr_msg = None
1249         self._nsr_regh = None
1250         self._key_pairs = key_pairs
1251         self._vlrs = []
1252         self._vnfrs = {}
1253         self._vnfds = {}
1254         self._vnffgrs = {}
1255         self._param_pools = {}
1256         self._scaling_groups = {}
1257         self._create_time = int(time.time())
1258         self._op_status = NetworkServiceStatus(dts, log, loop)
1259         self._config_status = NsrYang.ConfigStates.CONFIGURING
1260         self._config_status_details = None
1261         self._job_id = 0
1262         self.restart_mode = restart_mode
1263         self.config_store = rift.mano.config_data.config.ConfigStore(self._log)
1264         self._debug_running = False
1265         self._is_active = False
1266         self._vl_phase_completed = False
1267         self._vnf_phase_completed = False
1268         self.vlr_uptime_tasks = {}
1269
1270
1271         # Initalise the state to init
1272         # The NSR moves through the following transitions
1273         # 1. INIT -> VLS_READY once all the VLs in the NSD are created
1274         # 2. VLS_READY - VNFS_READY when all the VNFs in the NSD are created
1275         # 3. VNFS_READY - READY when the NSR is published
1276
1277         self.set_state(NetworkServiceRecordState.INIT)
1278
1279         self.substitute_input_parameters = InputParameterSubstitution(self._log)
1280
1281     @property
1282     def nsm_plugin(self):
1283         """ NSM Plugin """
1284         return self._nsm_plugin
1285
1286     def set_state(self, state):
1287         """ Set state for this NSR"""
1288         self._log.debug("Setting state to %s", state)
1289         # We are in init phase and is moving to the next state
1290         # The new state could be a FAILED state or VNF_INIIT_PHASE
1291         if self.state == NetworkServiceRecordState.VL_INIT_PHASE:
1292             self._vl_phase_completed = True
1293
1294         if self.state == NetworkServiceRecordState.VNF_INIT_PHASE:
1295             self._vnf_phase_completed = True
1296
1297         self._op_status.set_state(state)
1298
1299     @property
1300     def id(self):
1301         """ Get id for this NSR"""
1302         return self._nsr_cfg_msg.id
1303
1304     @property
1305     def name(self):
1306         """ Name of this network service record """
1307         return self._nsr_cfg_msg.name
1308
1309     @property
1310     def cloud_account_name(self):
1311         return self._nsr_cfg_msg.cloud_account
1312
1313     @property
1314     def om_datacenter_name(self):
1315         if self._nsr_cfg_msg.has_field('om_datacenter'):
1316             return self._nsr_cfg_msg.om_datacenter
1317         return None
1318
1319     @property
1320     def state(self):
1321         """State of this NetworkServiceRecord"""
1322         return self._op_status.state
1323
1324     @property
1325     def active(self):
1326         """ Is this NSR active ?"""
1327         return True if self._op_status.state == NetworkServiceRecordState.RUNNING else False
1328
1329     @property
1330     def vlrs(self):
1331         """ VLRs associated with this NSR"""
1332         return self._vlrs
1333
1334     @property
1335     def vnfrs(self):
1336         """ VNFRs associated with this NSR"""
1337         return self._vnfrs
1338
1339     @property
1340     def vnffgrs(self):
1341         """ VNFFGRs associated with this NSR"""
1342         return self._vnffgrs
1343
1344     @property
1345     def scaling_groups(self):
1346         """ Scaling groups associated with this NSR """
1347         return self._scaling_groups
1348
1349     @property
1350     def param_pools(self):
1351         """ Parameter value pools associated with this NSR"""
1352         return self._param_pools
1353
1354     @property
1355     def nsr_cfg_msg(self):
1356         return self._nsr_cfg_msg
1357
1358     @nsr_cfg_msg.setter
1359     def nsr_cfg_msg(self, msg):
1360         self._nsr_cfg_msg = msg
1361
1362     @property
1363     def nsd_msg(self):
1364         """ NSD Protobuf for this NSR """
1365         if self._nsd is not None:
1366             return self._nsd
1367         self._nsd = self._nsr_cfg_msg.nsd
1368         return self._nsd
1369
1370     @property
1371     def nsd_id(self):
1372         """ NSD ID for this NSR """
1373         return self.nsd_msg.id
1374
1375     @property
1376     def job_id(self):
1377         ''' Get a new job id for config primitive'''
1378         self._job_id += 1
1379         return self._job_id
1380
1381     @property
1382     def config_status(self):
1383         """ Config status for NSR """
1384         return self._config_status
1385
1386     def resolve_placement_group_cloud_construct(self, input_group):
1387         """
1388         Returns the cloud specific construct for placement group
1389         """
1390         copy_dict = ['name', 'requirement', 'strategy']
1391
1392         for group_info in self._nsr_cfg_msg.nsd_placement_group_maps:
1393             if group_info.placement_group_ref == input_group.name:
1394                 group = VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_PlacementGroupsInfo()
1395                 group_dict = {k:v for k,v in
1396                               group_info.as_dict().items() if k != 'placement_group_ref'}
1397                 for param in copy_dict:
1398                     group_dict.update({param: getattr(input_group, param)})
1399                 group.from_dict(group_dict)
1400                 return group
1401         return None
1402
1403
1404     def __str__(self):
1405         return "NSR(name={}, nsd_id={}, cloud_account={})".format(
1406                 self.name, self.nsd_id, self.cloud_account_name
1407                 )
1408
1409     def _get_vnfd(self, vnfd_id, config_xact):
1410         """  Fetch vnfd msg for the passed vnfd id """
1411         return self._nsm.get_vnfd(vnfd_id, config_xact)
1412
1413     def _get_vnfd_cloud_account(self, vnfd_member_index):
1414         """  Fetch Cloud Account for the passed vnfd id """
1415         if self._nsr_cfg_msg.vnf_cloud_account_map:
1416            vim_accounts = [(vnf.cloud_account,vnf.om_datacenter)  for vnf in self._nsr_cfg_msg.vnf_cloud_account_map \
1417                            if vnfd_member_index == vnf.member_vnf_index_ref]
1418            if vim_accounts and vim_accounts[0]:
1419                return vim_accounts[0]
1420         return (self.cloud_account_name,self.om_datacenter_name)
1421
1422     def _get_constituent_vnfd_msg(self, vnf_index):
1423         for const_vnfd in self.nsd_msg.constituent_vnfd:
1424             if const_vnfd.member_vnf_index == vnf_index:
1425                 return const_vnfd
1426
1427         raise ValueError("Constituent VNF index %s not found" % vnf_index)
1428
1429     def record_event(self, evt, evt_desc, evt_details=None, state=None):
1430         """ Record an event """
1431         self._op_status.record_event(evt, evt_desc, evt_details)
1432         if state is not None:
1433             self.set_state(state)
1434
1435     def scaling_trigger_str(self, trigger):
1436         SCALING_TRIGGER_STRS = {
1437             NsdYang.ScalingTrigger.PRE_SCALE_IN : 'pre-scale-in',
1438             NsdYang.ScalingTrigger.POST_SCALE_IN : 'post-scale-in',
1439             NsdYang.ScalingTrigger.PRE_SCALE_OUT : 'pre-scale-out',
1440             NsdYang.ScalingTrigger.POST_SCALE_OUT : 'post-scale-out',
1441         }
1442         try:
1443             return SCALING_TRIGGER_STRS[trigger]
1444         except Exception as e:
1445             self._log.error("Scaling trigger mapping error for {} : {}".
1446                             format(trigger, e))
1447             self._log.exception(e)
1448             return "Unknown trigger"
1449
1450     @asyncio.coroutine
1451     def instantiate_vls(self):
1452         """
1453         This function instantiates VLs for every VL in this Network Service
1454         """
1455         self._log.debug("Instantiating %d VLs in NSD id %s", len(self._vlrs),
1456                         self.id)
1457         for vlr in self._vlrs:
1458             yield from self.nsm_plugin.instantiate_vl(self, vlr)
1459             vlr.state = VlRecordState.ACTIVE
1460             self.vlr_uptime_tasks[vlr.id] = self._loop.create_task(self.vlr_uptime_update(vlr))
1461
1462
1463     def vlr_uptime_update(self, vlr):
1464         try:
1465
1466             vlr_ = RwVlrYang.YangData_Vlr_VlrCatalog_Vlr.from_dict({'id': vlr.id})
1467             while True:
1468                 vlr_.uptime = int(time.time()) - vlr._create_time
1469                 yield from self._vlr_handler.update(None, VirtualLinkRecord.vlr_xpath(vlr), vlr_)
1470                 yield from asyncio.sleep(2, loop=self._loop)
1471         except asyncio.CancelledError:
1472             self._log.debug("Received cancellation request for vlr_uptime_update task")
1473             yield from self._vlr_handler.delete(None, VirtualLinkRecord.vlr_xpath(vlr))
1474
1475
1476     @asyncio.coroutine
1477     def create(self, config_xact):
1478         """ Create this network service"""
1479         # Create virtual links  for all the external vnf
1480         # connection points in this NS
1481         yield from self.create_vls()
1482
1483         # Create VNFs in this network service
1484         yield from self.create_vnfs(config_xact)
1485
1486         # Create VNFFG for network service
1487         self.create_vnffgs()
1488
1489         # Create Scaling Groups for each scaling group in NSD
1490         self.create_scaling_groups()
1491
1492         # Create Parameter Pools
1493         self.create_param_pools()
1494
1495     @asyncio.coroutine
1496     def apply_scale_group_config_script(self, script, group, scale_instance, trigger, vnfrs=None):
1497         """ Apply config based on script for scale group """
1498
1499         @asyncio.coroutine
1500         def add_vnfrs_data(vnfrs_list):
1501             """ Add as a dict each of the VNFRs data """
1502             vnfrs_data = []
1503             for vnfr in vnfrs_list:
1504                 self._log.debug("Add VNFR {} data".format(vnfr))
1505                 vnfr_data = dict()
1506                 vnfr_data['name'] = vnfr.name
1507                 if trigger in [NsdYang.ScalingTrigger.PRE_SCALE_IN, NsdYang.ScalingTrigger.POST_SCALE_OUT]:
1508                     # Get VNF management and other IPs, etc
1509                     opdata = yield from self.fetch_vnfr(vnfr.xpath)
1510                     self._log.debug("VNFR {} op data: {}".format(vnfr.name, opdata))
1511                     try:
1512                         vnfr_data['rw_mgmt_ip'] = opdata.mgmt_interface.ip_address
1513                         vnfr_data['rw_mgmt_port'] = opdata.mgmt_interface.port
1514                     except Exception as e:
1515                         self._log.error("Unable to get management IP for vnfr {}:{}".
1516                                         format(vnfr.name, e))
1517
1518                     try:
1519                         vnfr_data['connection_points'] = []
1520                         for cp in opdata.connection_point:
1521                             con_pt = dict()
1522                             con_pt['name'] = cp.name
1523                             con_pt['ip_address'] = cp.ip_address
1524                             vnfr_data['connection_points'].append(con_pt)
1525                     except Exception as e:
1526                         self._log.error("Exception getting connections points for VNFR {}: {}".
1527                                         format(vnfr.name, e))
1528
1529                 vnfrs_data.append(vnfr_data)
1530                 self._log.debug("VNFRs data: {}".format(vnfrs_data))
1531
1532             return vnfrs_data
1533
1534         def add_nsr_data(nsr):
1535             nsr_data = dict()
1536             nsr_data['name'] = nsr.name
1537             return nsr_data
1538
1539         if script is None or len(script) == 0:
1540             self._log.error("Script not provided for scale group config: {}".format(group.name))
1541             return False
1542
1543         if script[0] == '/':
1544             path = script
1545         else:
1546             path = os.path.join(os.environ['RIFT_INSTALL'], "usr/bin", script)
1547         if not os.path.exists(path):
1548             self._log.error("Config faled for scale group {}: Script does not exist at {}".
1549                             format(group.name, path))
1550             return False
1551
1552         # Build a YAML file with all parameters for the script to execute
1553         # The data consists of 5 sections
1554         # 1. Trigger
1555         # 2. Scale group config
1556         # 3. VNFRs in the scale group
1557         # 4. VNFRs outside scale group
1558         # 5. NSR data
1559         data = dict()
1560         data['trigger'] = group.trigger_map(trigger)
1561         data['config'] = group.group_msg.as_dict()
1562
1563         if vnfrs:
1564             data["vnfrs_in_group"] = yield from add_vnfrs_data(vnfrs)
1565         else:
1566             data["vnfrs_in_group"] = yield from add_vnfrs_data(scale_instance.vnfrs)
1567
1568         data["vnfrs_others"] = yield from add_vnfrs_data(self.vnfrs.values())
1569         data["nsr"] = add_nsr_data(self)
1570
1571         tmp_file = None
1572         with tempfile.NamedTemporaryFile(delete=False) as tmp_file:
1573             tmp_file.write(yaml.dump(data, default_flow_style=True)
1574                     .encode("UTF-8"))
1575
1576         self._log.debug("Creating a temp file: {} with input data: {}".
1577                         format(tmp_file.name, data))
1578
1579         cmd = "{} {}".format(path, tmp_file.name)
1580         self._log.debug("Running the CMD: {}".format(cmd))
1581         proc = yield from asyncio.create_subprocess_shell(cmd, loop=self._loop)
1582         rc = yield from proc.wait()
1583         if rc:
1584             self._log.error("The script {} for scale group {} config returned: {}".
1585                             format(script, group.name, rc))
1586             return False
1587
1588         # Success
1589         return True
1590
1591
1592     @asyncio.coroutine
1593     def apply_scaling_group_config(self, trigger, group, scale_instance, vnfrs=None):
1594         """ Apply the config for the scaling group based on trigger """
1595         if group is None or scale_instance is None:
1596             return False
1597
1598         @asyncio.coroutine
1599         def update_config_status(success=True, err_msg=None):
1600             self._log.debug("Update %s config status to %r : %s",
1601                             scale_instance, success, err_msg)
1602             if (scale_instance.config_status == "failed"):
1603                 # Do not update the config status if it is already in failed state
1604                 return
1605
1606             if scale_instance.config_status == "configured":
1607                 # Update only to failed state an already configured scale instance
1608                 if not success:
1609                     scale_instance.config_status = "failed"
1610                     scale_instance.config_err_msg = err_msg
1611                     yield from self.update_state()
1612             else:
1613                 # We are in configuring state
1614                 # Only after post scale out mark instance as configured
1615                 if trigger == NsdYang.ScalingTrigger.POST_SCALE_OUT:
1616                     if success:
1617                         scale_instance.config_status = "configured"
1618                     else:
1619                         scale_instance.config_status = "failed"
1620                         scale_instance.config_err_msg = err_msg
1621                     yield from self.update_state()
1622
1623         config = group.trigger_config(trigger)
1624         if config is None:
1625             return True
1626
1627         self._log.debug("Scaling group {} config: {}".format(group.name, config))
1628         if config.has_field("ns_config_primitive_name_ref"):
1629             config_name = config.ns_config_primitive_name_ref
1630             nsd_msg = self.nsd_msg
1631             config_primitive = None
1632             for ns_cfg_prim in nsd_msg.service_primitive:
1633                 if ns_cfg_prim.name == config_name:
1634                     config_primitive = ns_cfg_prim
1635                     break
1636
1637             if config_primitive is None:
1638                 raise ValueError("Could not find ns_cfg_prim %s in nsr %s" % (config_name, self.name))
1639
1640             self._log.debug("Scaling group {} config primitive: {}".format(group.name, config_primitive))
1641             if config_primitive.has_field("user_defined_script"):
1642                 rc = yield from self.apply_scale_group_config_script(config_primitive.user_defined_script,
1643                                                                      group, scale_instance, trigger, vnfrs)
1644                 err_msg = None
1645                 if not rc:
1646                     err_msg = "Failed config for trigger {} using config script '{}'". \
1647                               format(self.scaling_trigger_str(trigger),
1648                                      config_primitive.user_defined_script)
1649                 yield from update_config_status(success=rc, err_msg=err_msg)
1650                 return rc
1651             else:
1652                 err_msg = "Failed config for trigger {} as config script is not specified". \
1653                           format(self.scaling_trigger_str(trigger))
1654                 yield from update_config_status(success=False, err_msg=err_msg)
1655                 raise NotImplementedError("Only script based config support for scale group for now: {}".
1656                                           format(group.name))
1657         else:
1658             err_msg = "Failed config for trigger {} as config primitive is not specified".\
1659                       format(self.scaling_trigger_str(trigger))
1660             yield from update_config_status(success=False, err_msg=err_msg)
1661             self._log.error("Config primitive not specified for config action in scale group %s" %
1662                             (group.name))
1663         return False
1664
1665     def create_scaling_groups(self):
1666         """ This function creates a NSScalingGroup for every scaling
1667         group defined in he NSD"""
1668
1669         for scaling_group_msg in self.nsd_msg.scaling_group_descriptor:
1670             self._log.debug("Found scaling_group %s in nsr id %s",
1671                             scaling_group_msg.name, self.id)
1672
1673             group_record = scale_group.ScalingGroup(
1674                     self._log,
1675                     scaling_group_msg
1676                     )
1677
1678             self._scaling_groups[group_record.name] = group_record
1679
1680     @asyncio.coroutine
1681     def create_scale_group_instance(self, group_name, index, config_xact, is_default=False):
1682         group = self._scaling_groups[group_name]
1683         scale_instance = group.create_instance(index, is_default)
1684
1685         @asyncio.coroutine
1686         def create_vnfs():
1687             self._log.debug("Creating %u VNFs associated with NS id %s scaling group %s",
1688                             len(self.nsd_msg.constituent_vnfd), self.id, self)
1689
1690             vnfrs = []
1691             for vnf_index, count in group.vnf_index_count_map.items():
1692                 const_vnfd_msg = self._get_constituent_vnfd_msg(vnf_index)
1693                 vnfd_msg = self._get_vnfd(const_vnfd_msg.vnfd_id_ref, config_xact)
1694
1695                 cloud_account_name, om_datacenter_name = self._get_vnfd_cloud_account(const_vnfd_msg.member_vnf_index)
1696                 if cloud_account_name is None:
1697                     cloud_account_name = self.cloud_account_name
1698                 for _ in range(count):
1699                     vnfr = yield from self.create_vnf_record(vnfd_msg, const_vnfd_msg, cloud_account_name, om_datacenter_name, group_name, index)
1700                     scale_instance.add_vnfr(vnfr)
1701                     vnfrs.append(vnfr)
1702
1703             return vnfrs
1704
1705         @asyncio.coroutine
1706         def instantiate_instance():
1707             self._log.debug("Creating %s VNFRS", scale_instance)
1708             vnfrs = yield from create_vnfs()
1709             yield from self.publish()
1710
1711             self._log.debug("Instantiating %s VNFRS for %s", len(vnfrs), scale_instance)
1712             scale_instance.operational_status = "vnf_init_phase"
1713             yield from self.update_state()
1714
1715             try:
1716                 rc = yield from self.apply_scaling_group_config(NsdYang.ScalingTrigger.PRE_SCALE_OUT,
1717                                                                 group, scale_instance, vnfrs)
1718                 if not rc:
1719                     self._log.error("Pre scale out config for scale group {} ({}) failed".
1720                                     format(group.name, index))
1721                     scale_instance.operational_status = "failed"
1722                 else:
1723                     yield from self.instantiate_vnfs(vnfrs)
1724
1725             except Exception as e:
1726                 self._log.exception("Failed to begin instantiatiation of vnfs for scale group {}: {}".
1727                                     format(group.name, e))
1728                 self._log.exception(e)
1729                 scale_instance.operational_status = "failed"
1730
1731             yield from self.update_state()
1732
1733         yield from instantiate_instance()
1734
1735     @asyncio.coroutine
1736     def delete_scale_group_instance(self, group_name, index):
1737         group = self._scaling_groups[group_name]
1738         scale_instance = group.get_instance(index)
1739         if scale_instance.is_default:
1740             raise ScalingOperationError("Cannot terminate a default scaling group instance")
1741
1742         scale_instance.operational_status = "terminate"
1743         yield from self.update_state()
1744
1745         @asyncio.coroutine
1746         def terminate_instance():
1747             self._log.debug("Terminating %s VNFRS" % scale_instance)
1748             rc = yield from self.apply_scaling_group_config(NsdYang.ScalingTrigger.PRE_SCALE_IN,
1749                                                             group, scale_instance)
1750             if not rc:
1751                 self._log.error("Pre scale in config for scale group {} ({}) failed".
1752                                 format(group.name, index))
1753
1754             # Going ahead with terminate, even if there is an error in pre-scale-in config
1755             # as this could be result of scale out failure and we need to cleanup this group
1756             yield from self.terminate_vnfrs(scale_instance.vnfrs)
1757             group.delete_instance(index)
1758
1759             scale_instance.operational_status = "vnf_terminate_phase"
1760             yield from self.update_state()
1761
1762         yield from terminate_instance()
1763
1764     @asyncio.coroutine
1765     def _update_scale_group_instances_status(self):
1766         @asyncio.coroutine
1767         def post_scale_out_task(group, instance):
1768             # Apply post scale out config once all VNFRs are active
1769             rc = yield from self.apply_scaling_group_config(NsdYang.ScalingTrigger.POST_SCALE_OUT,
1770                                                             group, instance)
1771             instance.operational_status = "running"
1772             if rc:
1773                 self._log.debug("Scale out for group {} and instance {} succeeded".
1774                                 format(group.name, instance.instance_id))
1775             else:
1776                 self._log.error("Post scale out config for scale group {} ({}) failed".
1777                                 format(group.name, instance.instance_id))
1778
1779             yield from self.update_state()
1780
1781         group_instances = {group: group.instances for group in self._scaling_groups.values()}
1782         for group, instances in group_instances.items():
1783             self._log.debug("Updating %s instance status", group)
1784             for instance in instances:
1785                 instance_vnf_state_list = [vnfr.state for vnfr in instance.vnfrs]
1786                 self._log.debug("Got vnfr instance states: %s", instance_vnf_state_list)
1787                 if instance.operational_status == "vnf_init_phase":
1788                     if all([state == VnfRecordState.ACTIVE for state in instance_vnf_state_list]):
1789                         instance.operational_status = "running"
1790
1791                         # Create a task for post scale out to allow us to sleep before attempting
1792                         # to configure newly created VM's
1793                         self._loop.create_task(post_scale_out_task(group, instance))
1794
1795                     elif any([state == VnfRecordState.FAILED for state in instance_vnf_state_list]):
1796                         self._log.debug("Scale out for group {} and instance {} failed".
1797                                         format(group.name, instance.instance_id))
1798                         instance.operational_status = "failed"
1799
1800                 elif instance.operational_status == "vnf_terminate_phase":
1801                     if all([state == VnfRecordState.TERMINATED for state in instance_vnf_state_list]):
1802                         instance.operational_status = "terminated"
1803                         rc = yield from self.apply_scaling_group_config(NsdYang.ScalingTrigger.POST_SCALE_IN,
1804                                                                          group, instance)
1805                         if rc:
1806                             self._log.debug("Scale in for group {} and instance {} succeeded".
1807                                             format(group.name, instance.instance_id))
1808                         else:
1809                             self._log.error("Post scale in config for scale group {} ({}) failed".
1810                                             format(group.name, instance.instance_id))
1811
1812     def create_vnffgs(self):
1813         """ This function creates VNFFGs for every VNFFG in the NSD
1814         associated with this NSR"""
1815
1816         for vnffgd in self.nsd_msg.vnffgd:
1817             self._log.debug("Found vnffgd %s in nsr id %s", vnffgd, self.id)
1818             vnffgr = VnffgRecord(self._dts,
1819                                  self._log,
1820                                  self._loop,
1821                                  self._nsm._vnffgmgr,
1822                                  self,
1823                                  self.name,
1824                                  vnffgd,
1825                                  self._sdn_account_name
1826                                  )
1827             self._vnffgrs[vnffgr.id] = vnffgr
1828
1829     def resolve_vld_ip_profile(self, nsd_msg, vld):
1830         self._log.debug("Receieved ip profile ref is %s",vld.ip_profile_ref)
1831         if not vld.has_field('ip_profile_ref'):
1832             return None
1833         profile = [profile for profile in nsd_msg.ip_profiles if profile.name == vld.ip_profile_ref]
1834         return profile[0] if profile else None
1835
1836     @asyncio.coroutine
1837     def _create_vls(self, vld, cloud_account,om_datacenter):
1838         """Create a VLR in the cloud account specified using the given VLD
1839
1840         Args:
1841             vld : VLD yang obj
1842             cloud_account : Cloud account name
1843
1844         Returns:
1845             VirtualLinkRecord
1846         """
1847         vlr = yield from VirtualLinkRecord.create_record(
1848                 self._dts,
1849                 self._log,
1850                 self._loop,
1851                 self.name,
1852                 vld,
1853                 cloud_account,
1854                 om_datacenter,
1855                 self.resolve_vld_ip_profile(self.nsd_msg, vld),
1856                 self.id,
1857                 restart_mode=self.restart_mode)
1858
1859         return vlr
1860
1861     def _extract_cloud_accounts_for_vl(self, vld):
1862         """
1863         Extracts the list of cloud accounts from the NS Config obj
1864
1865         Rules:
1866         1. Cloud accounts based connection point (vnf_cloud_account_map)
1867         Args:
1868             vld : VLD yang object
1869
1870         Returns:
1871             TYPE: Description
1872         """
1873         cloud_account_list = []
1874
1875         if self._nsr_cfg_msg.vnf_cloud_account_map:
1876             # Handle case where cloud_account is None
1877             vnf_cloud_map = {}
1878             for vnf in self._nsr_cfg_msg.vnf_cloud_account_map:
1879                 if vnf.cloud_account is not None or vnf.om_datacenter is not None:
1880                     vnf_cloud_map[vnf.member_vnf_index_ref] = (vnf.cloud_account,vnf.om_datacenter)
1881
1882             for vnfc in vld.vnfd_connection_point_ref:
1883                 cloud_account = vnf_cloud_map.get(
1884                         vnfc.member_vnf_index_ref,
1885                         (self.cloud_account_name,self.om_datacenter_name))
1886
1887                 cloud_account_list.append(cloud_account)
1888
1889         if self._nsr_cfg_msg.vl_cloud_account_map:
1890             for vld_map in self._nsr_cfg_msg.vl_cloud_account_map:
1891                 if vld_map.vld_id_ref == vld.id:
1892                     for cloud_account in vld_map.cloud_accounts:
1893                         cloud_account_list.extend((cloud_account,None))
1894                     for om_datacenter in vld_map.om_datacenters:
1895                         cloud_account_list.extend((None,om_datacenter))
1896
1897         # If no config has been provided then fall-back to the default
1898         # account
1899         if not cloud_account_list:
1900             cloud_account_list = [(self.cloud_account_name,self.om_datacenter_name)]
1901
1902         self._log.debug("VL {} cloud accounts: {}".
1903                         format(vld.name, cloud_account_list))
1904         return set(cloud_account_list)
1905
1906     @asyncio.coroutine
1907     def create_vls(self):
1908         """ This function creates VLs for every VLD in the NSD
1909         associated with this NSR"""
1910         for vld in self.nsd_msg.vld:
1911
1912             self._log.debug("Found vld %s in nsr id %s", vld, self.id)
1913             cloud_account_list = self._extract_cloud_accounts_for_vl(vld)
1914             for cloud_account,om_datacenter in cloud_account_list:
1915                 vlr = yield from self._create_vls(vld, cloud_account,om_datacenter)
1916                 self._vlrs.append(vlr)
1917
1918
1919     @asyncio.coroutine
1920     def create_vl_instance(self, vld):
1921         self._log.debug("Create VL for {}: {}".format(self.id, vld.as_dict()))
1922         # Check if the VL is already present
1923         vlr = None
1924         for vl in self._vlrs:
1925             if vl.vld_msg.id == vld.id:
1926                 self._log.debug("The VLD %s already in NSR %s as VLR %s with status %s",
1927                                 vld.id, self.id, vl.id, vl.state)
1928                 vlr = vl
1929                 if vlr.state != VlRecordState.TERMINATED:
1930                     err_msg = "VLR for VL %s in NSR %s already instantiated", \
1931                               vld, self.id
1932                     self._log.error(err_msg)
1933                     raise NsrVlUpdateError(err_msg)
1934                 break
1935
1936         if vlr is None:
1937             cloud_account_list = self._extract_cloud_accounts_for_vl(vld)
1938             for account,om_datacenter in cloud_account_list:
1939                 vlr = yield from self._create_vls(vld, account,om_datacenter)
1940                 self._vlrs.append(vlr)
1941
1942         vlr.state = VlRecordState.INSTANTIATION_PENDING
1943         yield from self.update_state()
1944
1945         try:
1946             yield from self.nsm_plugin.instantiate_vl(self, vlr)
1947             vlr.state = VlRecordState.ACTIVE
1948
1949         except Exception as e:
1950             err_msg = "Error instantiating VL for NSR {} and VLD {}: {}". \
1951                       format(self.id, vld.id, e)
1952             self._log.error(err_msg)
1953             self._log.exception(e)
1954             vlr.state = VlRecordState.FAILED
1955
1956         yield from self.update_state()
1957
1958     @asyncio.coroutine
1959     def delete_vl_instance(self, vld):
1960         for vlr in self._vlrs:
1961             if vlr.vld_msg.id == vld.id:
1962                 self._log.debug("Found VLR %s for VLD %s in NSR %s",
1963                                 vlr.id, vld.id, self.id)
1964                 vlr.state = VlRecordState.TERMINATE_PENDING
1965                 yield from self.update_state()
1966
1967                 try:
1968                     yield from self.nsm_plugin.terminate_vl(vlr)
1969                     vlr.state = VlRecordState.TERMINATED
1970                     self._vlrs.remove(vlr)
1971
1972                 except Exception as e:
1973                     err_msg = "Error terminating VL for NSR {} and VLD {}: {}". \
1974                               format(self.id, vld.id, e)
1975                     self._log.error(err_msg)
1976                     self._log.exception(e)
1977                     vlr.state = VlRecordState.FAILED
1978
1979                 yield from self.update_state()
1980                 break
1981
1982     @asyncio.coroutine
1983     def create_vnfs(self, config_xact):
1984         """
1985         This function creates VNFs for every VNF in the NSD
1986         associated with this NSR
1987         """
1988         self._log.debug("Creating %u VNFs associated with this NS id %s",
1989                         len(self.nsd_msg.constituent_vnfd), self.id)
1990
1991         for const_vnfd in self.nsd_msg.constituent_vnfd:
1992             if not const_vnfd.start_by_default:
1993                 self._log.debug("start_by_default set to False in constituent VNF (%s). Skipping start.",
1994                                 const_vnfd.member_vnf_index)
1995                 continue
1996
1997             vnfd_msg = self._get_vnfd(const_vnfd.vnfd_id_ref, config_xact)
1998             cloud_account_name,om_datacenter_name = self._get_vnfd_cloud_account(const_vnfd.member_vnf_index)
1999             if cloud_account_name is None:
2000                 cloud_account_name = self.cloud_account_name
2001             yield from self.create_vnf_record(vnfd_msg, const_vnfd, cloud_account_name, om_datacenter_name)
2002
2003
2004     def get_placement_groups(self, vnfd_msg, const_vnfd):
2005         placement_groups = []
2006         for group in self.nsd_msg.placement_groups:
2007             for member_vnfd in group.member_vnfd:
2008                 if (member_vnfd.vnfd_id_ref == vnfd_msg.id) and \
2009                    (member_vnfd.member_vnf_index_ref == const_vnfd.member_vnf_index):
2010                     group_info = self.resolve_placement_group_cloud_construct(group)
2011                     if group_info is None:
2012                         self._log.error("Could not resolve cloud-construct for placement group: %s", group.name)
2013                         ### raise PlacementGroupError("Could not resolve cloud-construct for placement group: {}".format(group.name))
2014                     else:
2015                         self._log.info("Successfully resolved cloud construct for placement group: %s for VNF: %s (Member Index: %s)",
2016                                        str(group_info),
2017                                        vnfd_msg.name,
2018                                        const_vnfd.member_vnf_index)
2019                         placement_groups.append(group_info)
2020         return placement_groups
2021
2022     @asyncio.coroutine
2023     def create_vnf_record(self, vnfd_msg, const_vnfd, cloud_account_name, om_datacenter_name, group_name=None, group_instance_id=None):
2024         # Fetch the VNFD associated with this VNF
2025         placement_groups = self.get_placement_groups(vnfd_msg, const_vnfd)
2026         self._log.info("Cloud Account for VNF %d is %s",const_vnfd.member_vnf_index,cloud_account_name)
2027         self._log.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s",
2028                        vnfd_msg.name,
2029                        const_vnfd.member_vnf_index,
2030                        [ group.name for group in placement_groups])
2031         vnfr = yield from VirtualNetworkFunctionRecord.create_record(self._dts,
2032                                             self._log,
2033                                             self._loop,
2034                                             vnfd_msg,
2035                                             const_vnfd,
2036                                             self.nsd_id,
2037                                             self.name,
2038                                             cloud_account_name,
2039                                             om_datacenter_name,
2040                                             self.id,
2041                                             group_name,
2042                                             group_instance_id,
2043                                             placement_groups,
2044                                             restart_mode=self.restart_mode,
2045                                             )
2046         if vnfr.id in self._vnfrs:
2047             err = "VNF with VNFR id %s already in vnf list" % (vnfr.id,)
2048             raise NetworkServiceRecordError(err)
2049
2050         self._vnfrs[vnfr.id] = vnfr
2051         self._nsm.vnfrs[vnfr.id] = vnfr
2052
2053         yield from vnfr.set_config_status(NsrYang.ConfigStates.INIT)
2054
2055         self._log.debug("Added VNFR %s to NSM VNFR list with id %s",
2056                         vnfr.name,
2057                         vnfr.id)
2058
2059         return vnfr
2060
2061     def create_param_pools(self):
2062         for param_pool in self.nsd_msg.parameter_pool:
2063             self._log.debug("Found parameter pool %s in nsr id %s", param_pool, self.id)
2064
2065             start_value = param_pool.range.start_value
2066             end_value = param_pool.range.end_value
2067             if end_value < start_value:
2068                 raise NetworkServiceRecordError(
2069                         "Parameter pool %s has invalid range (start: {}, end: {})".format(
2070                             start_value, end_value
2071                             )
2072                         )
2073
2074             self._param_pools[param_pool.name] = config_value_pool.ParameterValuePool(
2075                     self._log,
2076                     param_pool.name,
2077                     range(start_value, end_value)
2078                     )
2079
2080     @asyncio.coroutine
2081     def fetch_vnfr(self, vnfr_path):
2082         """ Fetch VNFR record """
2083         vnfr = None
2084         self._log.debug("Fetching VNFR with key %s while instantiating %s",
2085                         vnfr_path, self.id)
2086         res_iter = yield from self._dts.query_read(vnfr_path, rwdts.XactFlag.MERGE)
2087
2088         for ent in res_iter:
2089             res = yield from ent
2090             vnfr = res.result
2091
2092         return vnfr
2093
2094     @asyncio.coroutine
2095     def instantiate_vnfs(self, vnfrs):
2096         """
2097         This function instantiates VNFs for every VNF in this Network Service
2098         """
2099         self._log.debug("Instantiating %u VNFs in NS %s", len(vnfrs), self.id)
2100         for vnf in vnfrs:
2101             self._log.debug("Instantiating VNF: %s in NS %s", vnf, self.id)
2102             yield from self.nsm_plugin.instantiate_vnf(self, vnf)
2103
2104     @asyncio.coroutine
2105     def instantiate_vnffgs(self):
2106         """
2107         This function instantiates VNFFGs for every VNFFG in this Network Service
2108         """
2109         self._log.debug("Instantiating %u VNFFGs in NS %s",
2110                         len(self.nsd_msg.vnffgd), self.id)
2111         for _, vnfr in self.vnfrs.items():
2112             while vnfr.state in [VnfRecordState.INSTANTIATION_PENDING, VnfRecordState.INIT]:
2113                 self._log.debug("Received vnfr state for vnfr %s is %s; retrying",vnfr.name,vnfr.state)
2114                 yield from asyncio.sleep(2, loop=self._loop)
2115             if vnfr.state == VnfRecordState.ACTIVE:
2116                 self._log.debug("Received vnfr state for vnfr %s is %s ",vnfr.name,vnfr.state)
2117                 continue
2118             else:
2119                 self._log.debug("Received vnfr state for vnfr %s is %s; failing vnffg creation",vnfr.name,vnfr.state)
2120                 self._vnffgr_state = VnffgRecordState.FAILED
2121                 return
2122
2123         self._log.info("Waiting for 90 seconds for VMs to come up")
2124         yield from asyncio.sleep(90, loop=self._loop)
2125         self._log.info("Starting VNFFG orchestration")
2126         for vnffg in self._vnffgrs.values():
2127             self._log.debug("Instantiating VNFFG: %s in NS %s", vnffg, self.id)
2128             yield from vnffg.instantiate()
2129
2130     @asyncio.coroutine
2131     def instantiate_scaling_instances(self, config_xact):
2132         """ Instantiate any default scaling instances in this Network Service """
2133         for group in self._scaling_groups.values():
2134             for i in range(group.min_instance_count):
2135                 self._log.debug("Instantiating %s default scaling instance %s", group, i)
2136                 yield from self.create_scale_group_instance(
2137                         group.name, i, config_xact, is_default=True
2138                         )
2139
2140             for group_msg in self._nsr_cfg_msg.scaling_group:
2141                 if group_msg.scaling_group_name_ref != group.name:
2142                     continue
2143
2144                 for instance in group_msg.instance:
2145                     self._log.debug("Reloading %s scaling instance %s", group_msg, instance.id)
2146                     yield from self.create_scale_group_instance(
2147                             group.name, instance.id, config_xact, is_default=False
2148                             )
2149
2150     def has_scaling_instances(self):
2151         """ Return boolean indicating if the network service has default scaling groups """
2152         for group in self._scaling_groups.values():
2153             if group.min_instance_count > 0:
2154                 return True
2155
2156         for group_msg in self._nsr_cfg_msg.scaling_group:
2157             if len(group_msg.instance) > 0:
2158                 return True
2159
2160         return False
2161
2162     @asyncio.coroutine
2163     def publish(self):
2164         """ This function publishes this NSR """
2165         self._nsr_msg = self.create_msg()
2166
2167         self._log.debug("Publishing the NSR with xpath %s and nsr %s",
2168                         self.nsr_xpath,
2169                         self._nsr_msg)
2170
2171         if self._debug_running:
2172             self._log.debug("Publishing NSR in RUNNING state!")
2173             #raise()
2174
2175         with self._dts.transaction() as xact:
2176             yield from self._nsm.nsr_handler.update(xact, self.nsr_xpath, self._nsr_msg)
2177             if self._op_status.state == NetworkServiceRecordState.RUNNING:
2178                 self._debug_running = True
2179
2180     @asyncio.coroutine
2181     def unpublish(self, xact):
2182         """ Unpublish this NSR object """
2183         self._log.debug("Unpublishing Network service id %s", self.id)
2184         yield from self._nsm.nsr_handler.delete(xact, self.nsr_xpath)
2185
2186     @property
2187     def nsr_xpath(self):
2188         """ Returns the xpath associated with this NSR """
2189         return(
2190             "D,/nsr:ns-instance-opdata" +
2191             "/nsr:nsr[nsr:ns-instance-config-ref = '{}']"
2192             ).format(self.id)
2193
2194     @staticmethod
2195     def xpath_from_nsr(nsr):
2196         """ Returns the xpath associated with this NSR  op data"""
2197         return (NetworkServiceRecord.XPATH +
2198                 "[nsr:ns-instance-config-ref = '{}']").format(nsr.id)
2199
2200     @property
2201     def nsd_xpath(self):
2202         """ Return NSD config xpath."""
2203         return(
2204             "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}']"
2205             ).format(self.nsd_id)
2206
2207     @asyncio.coroutine
2208     def instantiate(self, config_xact):
2209         """"Instantiates a NetworkServiceRecord.
2210
2211         This function instantiates a Network service
2212         which involves the following steps,
2213
2214         * Instantiate every VL in NSD by sending create VLR request to DTS.
2215         * Instantiate every VNF in NSD by sending create VNF reuqest to DTS.
2216         * Publish the NSR details to DTS
2217
2218         Arguments:
2219             nsr:  The NSR configuration request containing nsr-id and nsd
2220             config_xact: The configuration transaction which initiated the instatiation
2221
2222         Raises:
2223             NetworkServiceRecordError if the NSR creation fails
2224
2225         Returns:
2226             No return value
2227         """
2228
2229         self._log.debug("Instantiating NS - %s xact - %s", self, config_xact)
2230
2231         # Move the state to INIITALIZING
2232         self.set_state(NetworkServiceRecordState.INIT)
2233
2234         event_descr = "Instantiation Request Received NSR Id:%s" % self.id
2235         self.record_event("instantiating", event_descr)
2236
2237         # Find the NSD
2238         self._nsd = self._nsr_cfg_msg.nsd
2239
2240         try:
2241             # Update ref count if nsd present in catalog
2242             self._nsm.get_nsd_ref(self.nsd_id)
2243
2244         except NetworkServiceDescriptorError:
2245             # This could be an NSD not in the nsd-catalog
2246             pass
2247
2248         # Merge any config and initial config primitive values
2249         self.config_store.merge_nsd_config(self.nsd_msg)
2250         self._log.debug("Merged NSD: {}".format(self.nsd_msg.as_dict()))
2251
2252         event_descr = "Fetched NSD with descriptor id %s" % self.nsd_id
2253         self.record_event("nsd-fetched", event_descr)
2254
2255         if self._nsd is None:
2256             msg = "Failed to fetch NSD with nsd-id [%s] for nsr-id %s"
2257             self._log.debug(msg, self.nsd_id, self.id)
2258             raise NetworkServiceRecordError(self)
2259
2260         self._log.debug("Got nsd result %s", self._nsd)
2261
2262         # Substitute any input parameters
2263         self.substitute_input_parameters(self._nsd, self._nsr_cfg_msg)
2264
2265         # Create the record
2266         yield from self.create(config_xact)
2267
2268         # Publish the NSR to DTS
2269         yield from self.publish()
2270
2271         @asyncio.coroutine
2272         def do_instantiate():
2273             """
2274                 Instantiate network service
2275             """
2276             self._log.debug("Instantiating VLs nsr id [%s] nsd id [%s]",
2277                             self.id, self.nsd_id)
2278
2279             # instantiate the VLs
2280             event_descr = ("Instantiating %s external VLs for NSR id %s" %
2281                            (len(self.nsd_msg.vld), self.id))
2282             self.record_event("begin-external-vls-instantiation", event_descr)
2283
2284             self.set_state(NetworkServiceRecordState.VL_INIT_PHASE)
2285
2286             yield from self.instantiate_vls()
2287
2288             # Publish the NSR to DTS
2289             yield from self.publish()
2290
2291             event_descr = ("Finished instantiating %s external VLs for NSR id %s" %
2292                            (len(self.nsd_msg.vld), self.id))
2293             self.record_event("end-external-vls-instantiation", event_descr)
2294
2295             self.set_state(NetworkServiceRecordState.VNF_INIT_PHASE)
2296
2297             self._log.debug("Instantiating VNFs  ...... nsr[%s], nsd[%s]",
2298                             self.id, self.nsd_id)
2299
2300             # instantiate the VNFs
2301             event_descr = ("Instantiating %s VNFS for NSR id %s" %
2302                            (len(self.nsd_msg.constituent_vnfd), self.id))
2303
2304             self.record_event("begin-vnf-instantiation", event_descr)
2305
2306             yield from self.instantiate_vnfs(self._vnfrs.values())
2307
2308             self._log.debug(" Finished instantiating %d VNFs for NSR id %s",
2309                             len(self.nsd_msg.constituent_vnfd), self.id)
2310
2311             event_descr = ("Finished instantiating %s VNFs for NSR id %s" %
2312                            (len(self.nsd_msg.constituent_vnfd), self.id))
2313             self.record_event("end-vnf-instantiation", event_descr)
2314
2315             if len(self.vnffgrs) > 0:
2316                 #self.set_state(NetworkServiceRecordState.VNFFG_INIT_PHASE)
2317                 event_descr = ("Instantiating %s VNFFGS for NSR id %s" %
2318                                (len(self.nsd_msg.vnffgd), self.id))
2319
2320                 self.record_event("begin-vnffg-instantiation", event_descr)
2321
2322                 yield from self.instantiate_vnffgs()
2323
2324                 event_descr = ("Finished instantiating %s VNFFGDs for NSR id %s" %
2325                                (len(self.nsd_msg.vnffgd), self.id))
2326                 self.record_event("end-vnffg-instantiation", event_descr)
2327
2328             if self.has_scaling_instances():
2329                 event_descr = ("Instantiating %s Scaling Groups for NSR id %s" %
2330                                (len(self._scaling_groups), self.id))
2331
2332                 self.record_event("begin-scaling-group-instantiation", event_descr)
2333                 yield from self.instantiate_scaling_instances(config_xact)
2334                 self.record_event("end-scaling-group-instantiation", event_descr)
2335
2336             # Give the plugin a chance to deploy the network service now that all
2337             # virtual links and vnfs are instantiated
2338             yield from self.nsm_plugin.deploy(self._nsr_msg)
2339
2340             self._log.debug("Publishing  NSR...... nsr[%s], nsd[%s]",
2341                             self.id, self.nsd_id)
2342
2343             # Publish the NSR to DTS
2344             yield from self.publish()
2345
2346             self._log.debug("Published  NSR...... nsr[%s], nsd[%s]",
2347                             self.id, self.nsd_id)
2348
2349         def on_instantiate_done(fut):
2350             # If the do_instantiate fails, then publish NSR with failed result
2351             if fut.exception() is not None:
2352                 self._log.error("NSR instantiation failed for NSR id %s: %s", self.id, str(fut.exception()))
2353                 self._loop.create_task(self.instantiation_failed(failed_reason=str(fut.exception())))
2354
2355         instantiate_task = self._loop.create_task(do_instantiate())
2356         instantiate_task.add_done_callback(on_instantiate_done)
2357
2358     @asyncio.coroutine
2359     def set_config_status(self, status, status_details=None):
2360         if self.config_status != status:
2361             self._log.debug("Updating NSR {} status for {} to {}".
2362                             format(self.name, self.config_status, status))
2363             self._config_status = status
2364             self._config_status_details = status_details
2365
2366             if self._config_status == NsrYang.ConfigStates.FAILED:
2367                 self.record_event("config-failed", "NS configuration failed",
2368                         evt_details=self._config_status_details)
2369
2370             yield from self.publish()
2371
2372     @asyncio.coroutine
2373     def is_active(self):
2374         """ This NS is active """
2375         self.set_state(NetworkServiceRecordState.RUNNING)
2376         if self._is_active:
2377             return
2378
2379         # Publish the NSR to DTS
2380         self._log.debug("Network service %s is active ", self.id)
2381         self._is_active = True
2382
2383         event_descr = "NSR in running state for NSR id %s" % self.id
2384         self.record_event("ns-running", event_descr)
2385
2386         yield from self.publish()
2387
2388     @asyncio.coroutine
2389     def instantiation_failed(self, failed_reason=None):
2390         """ The NS instantiation failed"""
2391         self._log.error("Network service id:%s, name:%s instantiation failed",
2392                         self.id, self.name)
2393         self.set_state(NetworkServiceRecordState.FAILED)
2394
2395         event_descr = "Instantiation of NS %s failed" % self.id
2396         self.record_event("ns-failed", event_descr, evt_details=failed_reason)
2397
2398         # Publish the NSR to DTS
2399         yield from self.publish()
2400
2401     @asyncio.coroutine
2402     def terminate_vnfrs(self, vnfrs):
2403         """ Terminate VNFRS in this network service """
2404         self._log.debug("Terminating VNFs in network service %s", self.id)
2405         for vnfr in vnfrs:
2406             yield from self.nsm_plugin.terminate_vnf(vnfr)
2407
2408     @asyncio.coroutine
2409     def terminate(self):
2410         """ Terminate a NetworkServiceRecord."""
2411         def terminate_vnffgrs():
2412             """ Terminate VNFFGRS in this network service """
2413             self._log.debug("Terminating VNFFGRs in network service %s", self.id)
2414             for vnffgr in self.vnffgrs.values():
2415                 yield from vnffgr.terminate()
2416
2417         def terminate_vlrs():
2418             """ Terminate VLRs in this netork service """
2419             self._log.debug("Terminating VLs in network service %s", self.id)
2420             for vlr in self.vlrs:
2421                 yield from self.nsm_plugin.terminate_vl(vlr)
2422                 vlr.state = VlRecordState.TERMINATED
2423                 if vlr.id in self.vlr_uptime_tasks:
2424                     self.vlr_uptime_tasks[vlr.id].cancel()
2425
2426         self._log.debug("Terminating network service id %s", self.id)
2427
2428         # Move the state to TERMINATE
2429         self.set_state(NetworkServiceRecordState.TERMINATE)
2430         event_descr = "Terminate being processed for NS Id:%s" % self.id
2431         self.record_event("terminate", event_descr)
2432
2433         # Move the state to VNF_TERMINATE_PHASE
2434         self._log.debug("Terminating VNFFGs in NS ID: %s", self.id)
2435         self.set_state(NetworkServiceRecordState.VNFFG_TERMINATE_PHASE)
2436         event_descr = "Terminating VNFFGS in NS Id:%s" % self.id
2437         self.record_event("terminating-vnffgss", event_descr)
2438         yield from terminate_vnffgrs()
2439
2440         # Move the state to VNF_TERMINATE_PHASE
2441         self.set_state(NetworkServiceRecordState.VNF_TERMINATE_PHASE)
2442         event_descr = "Terminating VNFS in NS Id:%s" % self.id
2443         self.record_event("terminating-vnfs", event_descr)
2444         yield from self.terminate_vnfrs(self.vnfrs.values())
2445
2446         # Move the state to VL_TERMINATE_PHASE
2447         self.set_state(NetworkServiceRecordState.VL_TERMINATE_PHASE)
2448         event_descr = "Terminating VLs in NS Id:%s" % self.id
2449         self.record_event("terminating-vls", event_descr)
2450         yield from terminate_vlrs()
2451
2452         yield from self.nsm_plugin.terminate_ns(self)
2453
2454         # Move the state to TERMINATED
2455         self.set_state(NetworkServiceRecordState.TERMINATED)
2456         event_descr = "Terminated NS Id:%s" % self.id
2457         self.record_event("terminated", event_descr)
2458
2459     def enable(self):
2460         """"Enable a NetworkServiceRecord."""
2461         pass
2462
2463     def disable(self):
2464         """"Disable a NetworkServiceRecord."""
2465         pass
2466
2467     def map_config_status(self):
2468         self._log.debug("Config status for ns {} is {}".
2469                         format(self.name, self._config_status))
2470         if self._config_status == NsrYang.ConfigStates.CONFIGURING:
2471             return 'configuring'
2472         if self._config_status == NsrYang.ConfigStates.FAILED:
2473             return 'failed'
2474         return 'configured'
2475
2476     def vl_phase_completed(self):
2477         """ Are VLs created in this NS?"""
2478         return self._vl_phase_completed
2479
2480     def vnf_phase_completed(self):
2481         """ Are VLs created in this NS?"""
2482         return self._vnf_phase_completed
2483
2484     def create_msg(self):
2485         """ The network serice record as a message """
2486         nsr_dict = {"ns_instance_config_ref": self.id}
2487         nsr = RwNsrYang.YangData_Nsr_NsInstanceOpdata_Nsr.from_dict(nsr_dict)
2488         #nsr.cloud_account = self.cloud_account_name
2489         nsr.sdn_account = self._sdn_account_name
2490         nsr.name_ref = self.name
2491         nsr.nsd_ref = self.nsd_id
2492         nsr.nsd_name_ref = self.nsd_msg.name
2493         nsr.operational_events = self._op_status.msg
2494         nsr.operational_status = self._op_status.yang_str()
2495         nsr.config_status = self.map_config_status()
2496         nsr.config_status_details = self._config_status_details
2497         nsr.create_time = self._create_time
2498         nsr.uptime = int(time.time()) - self._create_time
2499
2500         for cfg_prim in self.nsd_msg.service_primitive:
2501             cfg_prim = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_ServicePrimitive.from_dict(
2502                     cfg_prim.as_dict())
2503             nsr.service_primitive.append(cfg_prim)
2504
2505         for init_cfg in self.nsd_msg.initial_config_primitive:
2506             prim = NsrYang.NsrInitialConfigPrimitive.from_dict(
2507                 init_cfg.as_dict())
2508             nsr.initial_config_primitive.append(prim)
2509
2510         if self.vl_phase_completed():
2511             for vlr in self.vlrs:
2512                 nsr.vlr.append(vlr.create_nsr_vlr_msg(self.vnfrs.values()))
2513
2514         if self.vnf_phase_completed():
2515             for vnfr_id in self.vnfrs:
2516                 nsr.constituent_vnfr_ref.append(self.vnfrs[vnfr_id].const_vnfr_msg)
2517             for vnffgr in self.vnffgrs.values():
2518                 nsr.vnffgr.append(vnffgr.fetch_vnffgr())
2519             for scaling_group in self._scaling_groups.values():
2520                 nsr.scaling_group_record.append(scaling_group.create_record_msg())
2521
2522         return nsr
2523
2524     def all_vnfs_active(self):
2525         """ Are all VNFS in this NS active? """
2526         for _, vnfr in self.vnfrs.items():
2527             if vnfr.active is not True:
2528                 return False
2529         return True
2530
2531     @asyncio.coroutine
2532     def update_state(self):
2533         """ Re-evaluate this  NS's state """
2534         curr_state = self._op_status.state
2535
2536         if curr_state == NetworkServiceRecordState.TERMINATED:
2537             self._log.debug("NS (%s) in terminated state, not updating state", self.id)
2538             return
2539
2540         new_state = NetworkServiceRecordState.RUNNING
2541         self._log.info("Received update_state for nsr: %s, curr-state: %s",
2542                        self.id, curr_state)
2543
2544         # Check all the VNFRs are present
2545         for _, vnfr in self.vnfrs.items():
2546             if vnfr.state in [VnfRecordState.ACTIVE, VnfRecordState.TERMINATED]:
2547                 pass
2548             elif vnfr.state == VnfRecordState.FAILED:
2549                 if vnfr._prev_state != vnfr.state:
2550                     event_descr = "Instantiation of VNF %s failed" % vnfr.id
2551                     event_error_details = vnfr.state_failed_reason
2552                     self.record_event("vnf-failed", event_descr, evt_details=event_error_details)
2553                     vnfr.set_state(VnfRecordState.FAILED)
2554                 else:
2555                     self._log.info("VNF state did not change, curr=%s, prev=%s",
2556                                    vnfr.state, vnfr._prev_state)
2557                 new_state = NetworkServiceRecordState.FAILED
2558                 break
2559             else:
2560                 self._log.info("VNF %s in NSR %s is still not active; current state is: %s",
2561                                vnfr.id, self.id, vnfr.state)
2562                 new_state = curr_state
2563
2564         # If new state is RUNNING; check all VLs
2565         if new_state == NetworkServiceRecordState.RUNNING:
2566             for vl in self.vlrs:
2567
2568                 if vl.state in [VlRecordState.ACTIVE, VlRecordState.TERMINATED]:
2569                     pass
2570                 elif vl.state == VlRecordState.FAILED:
2571                     if vl.prev_state != vl.state:
2572                         event_descr = "Instantiation of VL %s failed" % vl.id
2573                         event_error_details = vl.state_failed_reason
2574                         self.record_event("vl-failed", event_descr, evt_details=event_error_details)
2575                         vl.prev_state = vl.state
2576                     else:
2577                         self._log.debug("VL %s already in failed state")
2578                 else:
2579                     if vl.state in [VlRecordState.INSTANTIATION_PENDING, VlRecordState.INIT]:
2580                         new_state = NetworkServiceRecordState.VL_INSTANTIATE
2581                         break
2582
2583                     if vl.state in [VlRecordState.TERMINATE_PENDING]:
2584                         new_state = NetworkServiceRecordState.VL_TERMINATE
2585                         break
2586
2587         # If new state is RUNNING; check VNFFGRs are also active
2588         if new_state == NetworkServiceRecordState.RUNNING:
2589             for _, vnffgr in self.vnffgrs.items():
2590                 self._log.info("Checking vnffgr state for nsr %s is: %s",
2591                                self.id, vnffgr.state)
2592                 if vnffgr.state == VnffgRecordState.ACTIVE:
2593                     pass
2594                 elif vnffgr.state == VnffgRecordState.FAILED:
2595                     event_descr = "Instantiation of VNFFGR %s failed" % vnffgr.id
2596                     self.record_event("vnffg-failed", event_descr)
2597                     new_state = NetworkServiceRecordState.FAILED
2598                     break
2599                 else:
2600                     self._log.info("VNFFGR %s in NSR %s is still not active; current state is: %s",
2601                                     vnffgr.id, self.id, vnffgr.state)
2602                     new_state = curr_state
2603
2604         # Update all the scaling group instance operational status to
2605         # reflect the state of all VNFR within that instance
2606         yield from self._update_scale_group_instances_status()
2607
2608         for _, group in self._scaling_groups.items():
2609             if group.state == scale_group.ScaleGroupState.SCALING_OUT:
2610                 new_state = NetworkServiceRecordState.SCALING_OUT
2611                 break
2612             elif group.state == scale_group.ScaleGroupState.SCALING_IN:
2613                 new_state = NetworkServiceRecordState.SCALING_IN
2614                 break
2615
2616         if new_state != curr_state:
2617             self._log.debug("Changing state of Network service %s from %s to %s",
2618                             self.id, curr_state, new_state)
2619             if new_state == NetworkServiceRecordState.RUNNING:
2620                 yield from self.is_active()
2621             elif new_state == NetworkServiceRecordState.FAILED:
2622                 # If the NS is already active and we entered scaling_in, scaling_out,
2623                 # do not mark the NS as failing if scaling operation failed.
2624                 if curr_state in [NetworkServiceRecordState.SCALING_OUT,
2625                                   NetworkServiceRecordState.SCALING_IN] and self._is_active:
2626                     new_state = NetworkServiceRecordState.RUNNING
2627                     self.set_state(new_state)
2628                 else:
2629                     yield from self.instantiation_failed()
2630             else:
2631                 self.set_state(new_state)
2632
2633         yield from self.publish()
2634
2635
2636 class InputParameterSubstitution(object):
2637     """
2638     This class is responsible for substituting input parameters into an NSD.
2639     """
2640
2641     def __init__(self, log):
2642         """Create an instance of InputParameterSubstitution
2643
2644         Arguments:
2645             log - a logger for this object to use
2646
2647         """
2648         self.log = log
2649
2650     def __call__(self, nsd, nsr_config):
2651         """Substitutes input parameters from the NSR config into the NSD
2652
2653         This call modifies the provided NSD with the input parameters that are
2654         contained in the NSR config.
2655
2656         Arguments:
2657             nsd        - a GI NSD object
2658             nsr_config - a GI NSR config object
2659
2660         """
2661         if nsd is None or nsr_config is None:
2662             return
2663
2664         # Create a lookup of the xpath elements that this descriptor allows
2665         # to be modified
2666         optional_input_parameters = set()
2667         for input_parameter in nsd.input_parameter_xpath:
2668             optional_input_parameters.add(input_parameter.xpath)
2669
2670         # Apply the input parameters to the descriptor
2671         if nsr_config.input_parameter:
2672             for param in nsr_config.input_parameter:
2673                 if param.xpath not in optional_input_parameters:
2674                     msg = "tried to set an invalid input parameter ({})"
2675                     self.log.error(msg.format(param.xpath))
2676                     continue
2677
2678                 self.log.debug(
2679                         "input-parameter:{} = {}".format(
2680                             param.xpath,
2681                             param.value,
2682                             )
2683                         )
2684
2685                 try:
2686                     xpath.setxattr(nsd, param.xpath, param.value)
2687
2688                 except Exception as e:
2689                     self.log.exception(e)
2690
2691
2692 class NetworkServiceDescriptor(object):
2693     """
2694     Network service descriptor class
2695     """
2696
2697     def __init__(self, dts, log, loop, nsd, nsm):
2698         self._dts = dts
2699         self._log = log
2700         self._loop = loop
2701
2702         self._nsd = nsd
2703         self._ref_count = 0
2704
2705         self._nsm = nsm
2706
2707     @property
2708     def id(self):
2709         """ Returns nsd id """
2710         return self._nsd.id
2711
2712     @property
2713     def name(self):
2714         """ Returns name of nsd """
2715         return self._nsd.name
2716
2717     @property
2718     def ref_count(self):
2719         """ Returns reference count"""
2720         return self._ref_count
2721
2722     def in_use(self):
2723         """ Returns whether nsd is in use or not """
2724         return True if self.ref_count > 0 else False
2725
2726     def ref(self):
2727         """ Take a reference on this object """
2728         self._ref_count += 1
2729
2730     def unref(self):
2731         """ Release reference on this object """
2732         if self.ref_count < 1:
2733             msg = ("Unref on a NSD object - nsd id %s, ref_count = %s" %
2734                    (self.id, self.ref_count))
2735             self._log.critical(msg)
2736             raise NetworkServiceDescriptorError(msg)
2737         self._ref_count -= 1
2738
2739     @property
2740     def msg(self):
2741         """ Return the message associated with this NetworkServiceDescriptor"""
2742         return self._nsd
2743
2744     @staticmethod
2745     def path_for_id(nsd_id):
2746         """ Return path for the passed nsd_id"""
2747         return "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}'".format(nsd_id)
2748
2749     def path(self):
2750         """ Return the message associated with this NetworkServiceDescriptor"""
2751         return NetworkServiceDescriptor.path_for_id(self.id)
2752
2753     def update(self, nsd):
2754         """ Update the NSD descriptor """
2755         self._nsd = nsd
2756
2757
2758 class NsdDtsHandler(object):
2759     """ The network service descriptor DTS handler """
2760     XPATH = "C,/nsd:nsd-catalog/nsd:nsd"
2761
2762     def __init__(self, dts, log, loop, nsm):
2763         self._dts = dts
2764         self._log = log
2765         self._loop = loop
2766         self._nsm = nsm
2767
2768         self._regh = None
2769
2770     @property
2771     def regh(self):
2772         """ Return registration handle """
2773         return self._regh
2774
2775     @asyncio.coroutine
2776     def register(self):
2777         """ Register for Nsd create/update/delete/read requests from dts """
2778
2779         def on_apply(dts, acg, xact, action, scratch):
2780             """Apply the  configuration"""
2781             is_recovery = xact.xact is None and action == rwdts.AppconfAction.INSTALL
2782             self._log.debug("Got nsd apply cfg (xact:%s) (action:%s)",
2783                             xact, action)
2784             # Create/Update an NSD record
2785             for cfg in self._regh.get_xact_elements(xact):
2786                 # Only interested in those NSD cfgs whose ID was received in prepare callback
2787                 if cfg.id in scratch.get('nsds', []) or is_recovery:
2788                     self._nsm.update_nsd(cfg)
2789
2790             scratch.pop('nsds', None)
2791
2792             return RwTypes.RwStatus.SUCCESS
2793
2794         @asyncio.coroutine
2795         def delete_nsd_libs(nsd_id):
2796             """ Remove any files uploaded with NSD and stored under $RIFT_ARTIFACTS/libs/<id> """
2797             try:
2798                 rift_artifacts_dir = os.environ['RIFT_ARTIFACTS']
2799                 nsd_dir = os.path.join(rift_artifacts_dir, 'launchpad/libs', nsd_id)
2800
2801                 if os.path.exists (nsd_dir):
2802                     shutil.rmtree(nsd_dir, ignore_errors=True)
2803             except Exception as e:
2804                 self._log.error("Exception in cleaning up NSD libs {}: {}".
2805                                 format(nsd_id, e))
2806                 self._log.excpetion(e)
2807
2808         @asyncio.coroutine
2809         def on_prepare(dts, acg, xact, xact_info, ks_path, msg, scratch):
2810             """ Prepare callback from DTS for NSD config """
2811
2812             self._log.info("Got nsd prepare - config received nsd id %s, msg %s",
2813                            msg.id, msg)
2814
2815             fref = ProtobufC.FieldReference.alloc()
2816             fref.goto_whole_message(msg.to_pbcm())
2817
2818             if fref.is_field_deleted():
2819                 # Delete an NSD record
2820                 self._log.debug("Deleting NSD with id %s", msg.id)
2821                 if self._nsm.nsd_in_use(msg.id):
2822                     self._log.debug("Cannot delete NSD in use - %s", msg.id)
2823                     err = "Cannot delete an NSD in use - %s" % msg.id
2824                     raise NetworkServiceDescriptorRefCountExists(err)
2825
2826                 yield from delete_nsd_libs(msg.id)
2827                 self._nsm.delete_nsd(msg.id)
2828             else:
2829                 # Add this NSD to scratch to create/update in apply callback
2830                 nsds = scratch.setdefault('nsds', [])
2831                 nsds.append(msg.id)
2832                 # acg._scratch['nsds'].append(msg.id)
2833
2834             xact_info.respond_xpath(rwdts.XactRspCode.ACK)
2835
2836         self._log.debug(
2837             "Registering for NSD config using xpath: %s",
2838             NsdDtsHandler.XPATH,
2839             )
2840
2841         acg_hdl = rift.tasklets.AppConfGroup.Handler(on_apply=on_apply)
2842         with self._dts.appconf_group_create(handler=acg_hdl) as acg:
2843             # Need a list in scratch to store NSDs to create/update later
2844             # acg._scratch['nsds'] = list()
2845             self._regh = acg.register(
2846                 xpath=NsdDtsHandler.XPATH,
2847                 flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY | rwdts.Flag.CACHE,
2848                 on_prepare=on_prepare)
2849
2850
2851 class VnfdDtsHandler(object):
2852     """ DTS handler for VNFD config changes """
2853     XPATH = "C,/vnfd:vnfd-catalog/vnfd:vnfd"
2854
2855     def __init__(self, dts, log, loop, nsm):
2856         self._dts = dts
2857         self._log = log
2858         self._loop = loop
2859         self._nsm = nsm
2860         self._regh = None
2861
2862     @property
2863     def regh(self):
2864         """ DTS registration handle """
2865         return self._regh
2866
2867     @asyncio.coroutine
2868     def register(self):
2869         """ Register for VNFD configuration"""
2870
2871         @asyncio.coroutine
2872         def on_apply(dts, acg, xact, action, scratch):
2873             """Apply the  configuration"""
2874             self._log.debug("Got NSM VNFD apply (xact: %s) (action: %s)(scr: %s)",
2875                             xact, action, scratch)
2876
2877             # Create/Update a VNFD record
2878             for cfg in self._regh.get_xact_elements(xact):
2879                 # Only interested in those VNFD cfgs whose ID was received in prepare callback
2880                 if cfg.id in scratch.get('vnfds', []):
2881                     self._nsm.update_vnfd(cfg)
2882
2883             for cfg in self._regh.elements:
2884                 if cfg.id in scratch.get('deleted_vnfds', []):
2885                     yield from self._nsm.delete_vnfd(cfg.id)
2886
2887             scratch.pop('vnfds', None)
2888             scratch.pop('deleted_vnfds', None)
2889
2890         @asyncio.coroutine
2891         def on_prepare(dts, acg, xact, xact_info, ks_path, msg, scratch):
2892             """ on prepare callback """
2893             self._log.debug("Got on prepare for VNFD (path: %s) (action: %s) (msg: %s)",
2894                             ks_path.to_xpath(RwNsmYang.get_schema()), xact_info.query_action, msg)
2895
2896             fref = ProtobufC.FieldReference.alloc()
2897             fref.goto_whole_message(msg.to_pbcm())
2898
2899             # Handle deletes in prepare_callback, but adds/updates in apply_callback
2900             if fref.is_field_deleted():
2901                 self._log.debug("Adding msg to deleted field")
2902                 deleted_vnfds = scratch.setdefault('deleted_vnfds', [])
2903                 deleted_vnfds.append(msg.id)
2904             else:
2905                 # Add this VNFD to scratch to create/update in apply callback
2906                 vnfds = scratch.setdefault('vnfds', [])
2907                 vnfds.append(msg.id)
2908
2909             xact_info.respond_xpath(rwdts.XactRspCode.ACK)
2910
2911         self._log.debug(
2912             "Registering for VNFD config using xpath: %s",
2913             VnfdDtsHandler.XPATH,
2914             )
2915         acg_hdl = rift.tasklets.AppConfGroup.Handler(on_apply=on_apply)
2916         with self._dts.appconf_group_create(handler=acg_hdl) as acg:
2917             # Need a list in scratch to store VNFDs to create/update later
2918             # acg._scratch['vnfds'] = list()
2919             # acg._scratch['deleted_vnfds'] = list()
2920             self._regh = acg.register(
2921                 xpath=VnfdDtsHandler.XPATH,
2922                 flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY,
2923                 on_prepare=on_prepare)
2924
2925 class NsrRpcDtsHandler(object):
2926     """ The network service instantiation RPC DTS handler """
2927     EXEC_NSR_CONF_XPATH = "I,/nsr:start-network-service"
2928     EXEC_NSR_CONF_O_XPATH = "O,/nsr:start-network-service"
2929     NETCONF_IP_ADDRESS = "127.0.0.1"
2930     NETCONF_PORT = 2022
2931     RESTCONF_PORT = 8888
2932     NETCONF_USER = "admin"
2933     NETCONF_PW = "admin"
2934     REST_BASE_V2_URL = 'https://{}:{}/v2/api/'.format("127.0.0.1",8888)
2935
2936     def __init__(self, dts, log, loop, nsm):
2937         self._dts = dts
2938         self._log = log
2939         self._loop = loop
2940         self._nsm = nsm
2941         self._nsd = None
2942
2943         self._ns_regh = None
2944
2945         self._manager = None
2946         self._nsr_config_url = NsrRpcDtsHandler.REST_BASE_V2_URL + 'config/ns-instance-config'
2947
2948         self._model = RwYang.Model.create_libncx()
2949         self._model.load_schema_ypbc(RwNsrYang.get_schema())
2950
2951     @property
2952     def nsm(self):
2953         """ Return the NS manager instance """
2954         return self._nsm
2955
2956     @staticmethod
2957     def wrap_netconf_config_xml(xml):
2958         xml = '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml)
2959         return xml
2960
2961     @asyncio.coroutine
2962     def _connect(self, timeout_secs=240):
2963
2964         start_time = time.time()
2965         while (time.time() - start_time) < timeout_secs:
2966
2967             try:
2968                 self._log.debug("Attemping NsmTasklet netconf connection.")
2969
2970                 manager = yield from ncclient.asyncio_manager.asyncio_connect(
2971                         loop=self._loop,
2972                         host=NsrRpcDtsHandler.NETCONF_IP_ADDRESS,
2973                         port=NsrRpcDtsHandler.NETCONF_PORT,
2974                         username=NsrRpcDtsHandler.NETCONF_USER,
2975                         password=NsrRpcDtsHandler.NETCONF_PW,
2976                         allow_agent=False,
2977                         look_for_keys=False,
2978                         hostkey_verify=False,
2979                         )
2980
2981                 return manager
2982
2983             except ncclient.transport.errors.SSHError as e:
2984                 self._log.warning("Netconf connection to launchpad %s failed: %s",
2985                                   NsrRpcDtsHandler.NETCONF_IP_ADDRESS, str(e))
2986
2987             yield from asyncio.sleep(5, loop=self._loop)
2988
2989         raise NsrInstantiationFailed("Failed to connect to Launchpad within %s seconds" %
2990                                       timeout_secs)
2991
2992     def _apply_ns_instance_config(self,payload_dict):
2993         #self._log.debug("At apply NS instance config with payload %s",payload_dict)
2994         req_hdr= {'accept':'application/vnd.yang.data+json','content-type':'application/vnd.yang.data+json'}
2995         response=requests.post(self._nsr_config_url, headers=req_hdr, auth=('admin', 'admin'),data=payload_dict,verify=False)
2996         return response
2997
2998     @asyncio.coroutine
2999     def register(self):
3000         """ Register for NS monitoring read from dts """
3001         @asyncio.coroutine
3002         def on_ns_config_prepare(xact_info, action, ks_path, msg):
3003             """ prepare callback from dts start-network-service"""
3004             assert action == rwdts.QueryAction.RPC
3005             rpc_ip = msg
3006             rpc_op = NsrYang.YangOutput_Nsr_StartNetworkService.from_dict({
3007                     "nsr_id":str(uuid.uuid4())
3008                 })
3009
3010             if not ('name' in rpc_ip and  'nsd_ref' in rpc_ip and ('cloud_account' in rpc_ip or 'om_datacenter' in rpc_ip)):
3011                 self._log.error("Mandatory parameters name or nsd_ref or cloud account not found in start-network-service {}".format(rpc_ip))
3012
3013
3014             self._log.debug("start-network-service RPC input: {}".format(rpc_ip))
3015
3016             try:
3017                 # Add used value to the pool
3018                 self._log.debug("RPC output: {}".format(rpc_op))
3019
3020                 nsd_copy = self.nsm.get_nsd(rpc_ip.nsd_ref)
3021
3022                 #if not self._manager:
3023                 #    self._manager = yield from self._connect()
3024
3025                 self._log.debug("Configuring ns-instance-config with name  %s nsd-ref: %s",
3026                         rpc_ip.name, rpc_ip.nsd_ref)
3027
3028                 ns_instance_config_dict = {"id":rpc_op.nsr_id, "admin_status":"ENABLED"}
3029                 ns_instance_config_copy_dict = {k:v for k, v in rpc_ip.as_dict().items()
3030                                                 if k in RwNsrYang.YangData_Nsr_NsInstanceConfig_Nsr().fields}
3031                 ns_instance_config_dict.update(ns_instance_config_copy_dict)
3032
3033                 ns_instance_config = RwNsrYang.YangData_Nsr_NsInstanceConfig_Nsr.from_dict(ns_instance_config_dict)
3034                 ns_instance_config.nsd = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_Nsd()
3035                 ns_instance_config.nsd.from_dict(nsd_copy.msg.as_dict())
3036
3037                 payload_dict = ns_instance_config.to_json(self._model)
3038                 #xml = ns_instance_config.to_xml_v2(self._model)
3039                 #netconf_xml = self.wrap_netconf_config_xml(xml)
3040
3041                 #self._log.debug("Sending configure ns-instance-config xml to %s: %s",
3042                 #        netconf_xml, NsrRpcDtsHandler.NETCONF_IP_ADDRESS)
3043                 self._log.debug("Sending configure ns-instance-config json to %s: %s",
3044                         self._nsr_config_url,ns_instance_config)
3045
3046                 #response = yield from self._manager.edit_config(
3047                 #           target="running",
3048                 #           config=netconf_xml,
3049                 #           )
3050                 response = yield from self._loop.run_in_executor(
3051                     None,
3052                     self._apply_ns_instance_config,
3053                     payload_dict
3054                     )
3055                 response.raise_for_status()
3056                 self._log.debug("Received edit config response: %s", response.json())
3057
3058                 xact_info.respond_xpath(rwdts.XactRspCode.ACK,
3059                                         NsrRpcDtsHandler.EXEC_NSR_CONF_O_XPATH,
3060                                         rpc_op)
3061             except Exception as e:
3062                 self._log.error("Exception processing the "
3063                                 "start-network-service: {}".format(e))
3064                 self._log.exception(e)
3065                 xact_info.respond_xpath(rwdts.XactRspCode.NACK,
3066                                         NsrRpcDtsHandler.EXEC_NSR_CONF_O_XPATH)
3067
3068
3069         hdl_ns = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_ns_config_prepare,)
3070
3071         with self._dts.group_create() as group:
3072             self._ns_regh = group.register(xpath=NsrRpcDtsHandler.EXEC_NSR_CONF_XPATH,
3073                                            handler=hdl_ns,
3074                                            flags=rwdts.Flag.PUBLISHER,
3075                                           )
3076
3077
3078 class NsrDtsHandler(object):
3079     """ The network service DTS handler """
3080     NSR_XPATH = "C,/nsr:ns-instance-config/nsr:nsr"
3081     SCALE_INSTANCE_XPATH = "C,/nsr:ns-instance-config/nsr:nsr/nsr:scaling-group/nsr:instance"
3082     KEY_PAIR_XPATH = "C,/nsr:key-pair"
3083
3084     def __init__(self, dts, log, loop, nsm):
3085         self._dts = dts
3086         self._log = log
3087         self._loop = loop
3088         self._nsm = nsm
3089
3090         self._nsr_regh = None
3091         self._scale_regh = None
3092         self._key_pair_regh = None
3093
3094     @property
3095     def nsm(self):
3096         """ Return the NS manager instance """
3097         return self._nsm
3098
3099     @asyncio.coroutine
3100     def register(self):
3101         """ Register for Nsr create/update/delete/read requests from dts """
3102
3103         def nsr_id_from_keyspec(ks):
3104             nsr_path_entry = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr.schema().keyspec_to_entry(ks)
3105             nsr_id = nsr_path_entry.key00.id
3106             return nsr_id
3107
3108         def group_name_from_keyspec(ks):
3109             group_path_entry = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup.schema().keyspec_to_entry(ks)
3110             group_name = group_path_entry.key00.scaling_group_name_ref
3111             return group_name
3112
3113         def is_instance_in_reg_elements(nsr_id, group_name, instance_id):
3114             """ Return boolean indicating if scaling group instance was already commited previously.
3115
3116             By looking at the existing elements in this registration handle (elements not part
3117             of this current xact), we can tell if the instance was configured previously without
3118             keeping any application state.
3119             """
3120             for instance_cfg, keyspec in self._nsr_regh.get_xact_elements(include_keyspec=True):
3121                 elem_nsr_id = nsr_id_from_keyspec(keyspec)
3122                 elem_group_name = group_name_from_keyspec(keyspec)
3123
3124                 if elem_nsr_id != nsr_id or group_name != elem_group_name:
3125                     continue
3126
3127                 if instance_cfg.id == instance_id:
3128                     return True
3129
3130             return False
3131
3132         def get_scale_group_instance_delta(nsr_id, group_name, xact):
3133             delta = {"added": [], "deleted": []}
3134             for instance_cfg, keyspec in self._scale_regh.get_xact_elements(xact, include_keyspec=True):
3135                 elem_nsr_id = nsr_id_from_keyspec(keyspec)
3136                 if elem_nsr_id != nsr_id:
3137                     continue
3138
3139                 elem_group_name = group_name_from_keyspec(keyspec)
3140                 if elem_group_name != group_name:
3141                     continue
3142
3143                 delta["added"].append(instance_cfg.id)
3144
3145             for instance_cfg, keyspec in self._scale_regh.get_xact_elements(include_keyspec=True):
3146                 elem_nsr_id = nsr_id_from_keyspec(keyspec)
3147                 if elem_nsr_id != nsr_id:
3148                     continue
3149
3150                 elem_group_name = group_name_from_keyspec(keyspec)
3151                 if elem_group_name != group_name:
3152                     continue
3153
3154                 if instance_cfg.id in delta["added"]:
3155                     delta["added"].remove(instance_cfg.id)
3156                 else:
3157                     delta["deleted"].append(instance_cfg.id)
3158
3159             return delta
3160
3161         @asyncio.coroutine
3162         def update_nsr_nsd(nsr_id, xact, scratch):
3163
3164             @asyncio.coroutine
3165             def get_nsr_vl_delta(nsr_id, xact, scratch):
3166                 delta = {"added": [], "deleted": []}
3167                 for instance_cfg, keyspec in self._nsr_regh.get_xact_elements(xact, include_keyspec=True):
3168                     elem_nsr_id = nsr_id_from_keyspec(keyspec)
3169                     if elem_nsr_id != nsr_id:
3170                         continue
3171
3172                     if 'vld' in instance_cfg.nsd:
3173                         for vld in instance_cfg.nsd.vld:
3174                             delta["added"].append(vld)
3175
3176                 for instance_cfg, keyspec in self._nsr_regh.get_xact_elements(include_keyspec=True):
3177                     self._log.debug("NSR update: %s", instance_cfg)
3178                     elem_nsr_id = nsr_id_from_keyspec(keyspec)
3179                     if elem_nsr_id != nsr_id:
3180                         continue
3181
3182                     if 'vld' in instance_cfg.nsd:
3183                         for vld in instance_cfg.nsd.vld:
3184                             if vld in delta["added"]:
3185                                 delta["added"].remove(vld)
3186                             else:
3187                                 delta["deleted"].append(vld)
3188
3189                 return delta
3190
3191             vl_delta = yield from get_nsr_vl_delta(nsr_id, xact, scratch)
3192             self._log.debug("Got NSR:%s VL instance delta: %s", nsr_id, vl_delta)
3193
3194             for vld in vl_delta["added"]:
3195                 yield from self._nsm.nsr_instantiate_vl(nsr_id, vld)
3196
3197             for vld in vl_delta["deleted"]:
3198                 yield from self._nsm.nsr_terminate_vl(nsr_id, vld)
3199
3200         def get_add_delete_update_cfgs(dts_member_reg, xact, key_name, scratch):
3201             # Unfortunately, it is currently difficult to figure out what has exactly
3202             # changed in this xact without Pbdelta support (RIFT-4916)
3203             # As a workaround, we can fetch the pre and post xact elements and
3204             # perform a comparison to figure out adds/deletes/updates
3205             xact_cfgs = list(dts_member_reg.get_xact_elements(xact))
3206             curr_cfgs = list(dts_member_reg.elements)
3207
3208             xact_key_map = {getattr(cfg, key_name): cfg for cfg in xact_cfgs}
3209             curr_key_map = {getattr(cfg, key_name): cfg for cfg in curr_cfgs}
3210
3211             # Find Adds
3212             added_keys = set(xact_key_map) - set(curr_key_map)
3213             added_cfgs = [xact_key_map[key] for key in added_keys]
3214
3215             # Find Deletes
3216             deleted_keys = set(curr_key_map) - set(xact_key_map)
3217             deleted_cfgs = [curr_key_map[key] for key in deleted_keys]
3218
3219             # Find Updates
3220             updated_keys = set(curr_key_map) & set(xact_key_map)
3221             updated_cfgs = [xact_key_map[key] for key in updated_keys
3222                             if xact_key_map[key] != curr_key_map[key]]
3223
3224             return added_cfgs, deleted_cfgs, updated_cfgs
3225
3226         def get_nsr_key_pairs(dts_member_reg, xact):
3227             key_pairs = {}
3228             for instance_cfg, keyspec in dts_member_reg.get_xact_elements(xact, include_keyspec=True):
3229                 self._log.debug("Key pair received is {} KS: {}".format(instance_cfg, keyspec))
3230                 xpath = keyspec.to_xpath(RwNsrYang.get_schema())
3231                 key_pairs[instance_cfg.name] = instance_cfg
3232             return key_pairs
3233
3234         def on_apply(dts, acg, xact, action, scratch):
3235             """Apply the  configuration"""
3236             self._log.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3237                             xact, action, scratch)
3238
3239             def handle_create_nsr(msg, key_pairs=None, restart_mode=False):
3240                 # Handle create nsr requests """
3241                 # Do some validations
3242                 if not msg.has_field("nsd"):
3243                     err = "NSD not provided"
3244                     self._log.error(err)
3245                     raise NetworkServiceRecordError(err)
3246
3247                 self._log.debug("Creating NetworkServiceRecord %s  from nsr config  %s",
3248                                msg.id, msg.as_dict())
3249                 nsr = self.nsm.create_nsr(msg, key_pairs=key_pairs, restart_mode=restart_mode)
3250                 return nsr
3251
3252             def handle_delete_nsr(msg):
3253                 @asyncio.coroutine
3254                 def delete_instantiation(ns_id):
3255                     """ Delete instantiation """
3256                     with self._dts.transaction() as xact:
3257                         yield from self._nsm.terminate_ns(ns_id, xact)
3258
3259                 # Handle delete NSR requests
3260                 self._log.info("Delete req for  NSR Id: %s received", msg.id)
3261                 # Terminate the NSR instance
3262                 nsr = self._nsm.get_ns_by_nsr_id(msg.id)
3263
3264                 nsr.set_state(NetworkServiceRecordState.TERMINATE_RCVD)
3265                 event_descr = "Terminate rcvd for NS Id:%s" % msg.id
3266                 nsr.record_event("terminate-rcvd", event_descr)
3267
3268                 self._loop.create_task(delete_instantiation(msg.id))
3269
3270             @asyncio.coroutine
3271             def begin_instantiation(nsr):
3272                 # Begin instantiation
3273                 self._log.info("Beginning NS instantiation: %s", nsr.id)
3274                 yield from self._nsm.instantiate_ns(nsr.id, xact)
3275
3276             self._log.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3277                             xact, action, scratch)
3278
3279             if action == rwdts.AppconfAction.INSTALL and xact.id is None:
3280                 key_pairs = []
3281                 for element in self._key_pair_regh.elements:
3282                     key_pairs.append(element)
3283                 for element in self._nsr_regh.elements:
3284                     nsr = handle_create_nsr(element, key_pairs, restart_mode=True)
3285                     self._loop.create_task(begin_instantiation(nsr))
3286
3287
3288             (added_msgs, deleted_msgs, updated_msgs) = get_add_delete_update_cfgs(self._nsr_regh,
3289                                                                                   xact,
3290                                                                                   "id",
3291                                                                                   scratch)
3292             self._log.debug("Added: %s, Deleted: %s, Updated: %s", added_msgs,
3293                             deleted_msgs, updated_msgs)
3294
3295             for msg in added_msgs:
3296                 if msg.id not in self._nsm.nsrs:
3297                     self._log.info("Create NSR received in on_apply to instantiate NS:%s", msg.id)
3298                     key_pairs = get_nsr_key_pairs(self._key_pair_regh, xact)
3299                     nsr = handle_create_nsr(msg,key_pairs)
3300                     self._loop.create_task(begin_instantiation(nsr))
3301
3302             for msg in deleted_msgs:
3303                 self._log.info("Delete NSR received in on_apply to terminate NS:%s", msg.id)
3304                 try:
3305                     handle_delete_nsr(msg)
3306                 except Exception:
3307                     self._log.exception("Failed to terminate NS:%s", msg.id)
3308
3309             for msg in updated_msgs:
3310                 self._log.info("Update NSR received in on_apply: %s", msg)
3311
3312                 self._nsm.nsr_update_cfg(msg.id, msg)
3313
3314                 if 'nsd' in msg:
3315                     self._loop.create_task(update_nsr_nsd(msg.id, xact, scratch))
3316
3317                 for group in msg.scaling_group:
3318                     instance_delta = get_scale_group_instance_delta(msg.id, group.scaling_group_name_ref, xact)
3319                     self._log.debug("Got NSR:%s scale group instance delta: %s", msg.id, instance_delta)
3320
3321                     for instance_id in instance_delta["added"]:
3322                         self._nsm.scale_nsr_out(msg.id, group.scaling_group_name_ref, instance_id, xact)
3323
3324                     for instance_id in instance_delta["deleted"]:
3325                         self._nsm.scale_nsr_in(msg.id, group.scaling_group_name_ref, instance_id)
3326
3327
3328             return RwTypes.RwStatus.SUCCESS
3329
3330         @asyncio.coroutine
3331         def on_prepare(dts, acg, xact, xact_info, ks_path, msg, scratch):
3332             """ Prepare calllback from DTS for NSR """
3333
3334             xpath = ks_path.to_xpath(RwNsrYang.get_schema())
3335             action = xact_info.query_action
3336             self._log.debug(
3337                     "Got Nsr prepare callback (xact: %s) (action: %s) (info: %s), %s:%s)",
3338                     xact, action, xact_info, xpath, msg
3339                     )
3340
3341             @asyncio.coroutine
3342             def delete_instantiation(ns_id):
3343                 """ Delete instantiation """
3344                 yield from self._nsm.terminate_ns(ns_id, None)
3345
3346             def handle_delete_nsr():
3347                 """ Handle delete NSR requests """
3348                 self._log.info("Delete req for  NSR Id: %s received", msg.id)
3349                 # Terminate the NSR instance
3350                 nsr = self._nsm.get_ns_by_nsr_id(msg.id)
3351
3352                 nsr.set_state(NetworkServiceRecordState.TERMINATE_RCVD)
3353                 event_descr = "Terminate rcvd for NS Id:%s" % msg.id
3354                 nsr.record_event("terminate-rcvd", event_descr)
3355
3356                 self._loop.create_task(delete_instantiation(msg.id))
3357
3358             fref = ProtobufC.FieldReference.alloc()
3359             fref.goto_whole_message(msg.to_pbcm())
3360
3361             if action in [rwdts.QueryAction.CREATE, rwdts.QueryAction.UPDATE, rwdts.QueryAction.DELETE]:
3362                 # if this is an NSR create
3363                 if action != rwdts.QueryAction.DELETE and msg.id not in self._nsm.nsrs:
3364                     # Ensure the Cloud account/datacenter has been specified
3365                     if not msg.has_field("cloud_account") and not msg.has_field("om_datacenter"):
3366                         raise NsrInstantiationFailed("Cloud account or datacenter not specified in NSR")
3367
3368                     # Check if nsd is specified
3369                     if not msg.has_field("nsd"):
3370                         raise NsrInstantiationFailed("NSD not specified in NSR")
3371
3372                 else:
3373                     nsr = self._nsm.nsrs[msg.id]
3374
3375                     if msg.has_field("nsd"):
3376                         if nsr.state != NetworkServiceRecordState.RUNNING:
3377                             raise NsrVlUpdateError("Unable to update VL when NSR not in running state")
3378                         if 'vld' not in msg.nsd or len(msg.nsd.vld) == 0:
3379                             raise NsrVlUpdateError("NS config NSD should have atleast 1 VLD defined")
3380
3381                     if msg.has_field("scaling_group"):
3382                         if nsr.state != NetworkServiceRecordState.RUNNING:
3383                             raise ScalingOperationError("Unable to perform scaling action when NS is not in running state")
3384
3385                         if len(msg.scaling_group) > 1:
3386                             raise ScalingOperationError("Only a single scaling group can be configured at a time")
3387
3388                         for group_msg in msg.scaling_group:
3389                             num_new_group_instances = len(group_msg.instance)
3390                             if num_new_group_instances > 1:
3391                                 raise ScalingOperationError("Only a single scaling instance can be modified at a time")
3392
3393                             elif num_new_group_instances == 1:
3394                                 scale_group = nsr.scaling_groups[group_msg.scaling_group_name_ref]
3395                                 if action in [rwdts.QueryAction.CREATE, rwdts.QueryAction.UPDATE]:
3396                                     if len(scale_group.instances) == scale_group.max_instance_count:
3397                                         raise ScalingOperationError("Max instances for %s reached" % scale_group)
3398
3399             acg.handle.prepare_complete_ok(xact_info.handle)
3400
3401
3402         self._log.debug("Registering for NSR config using xpath: %s",
3403                         NsrDtsHandler.NSR_XPATH)
3404
3405         acg_hdl = rift.tasklets.AppConfGroup.Handler(on_apply=on_apply)
3406         with self._dts.appconf_group_create(handler=acg_hdl) as acg:
3407             self._nsr_regh = acg.register(xpath=NsrDtsHandler.NSR_XPATH,
3408                                       flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY | rwdts.Flag.CACHE,
3409                                       on_prepare=on_prepare)
3410
3411             self._scale_regh = acg.register(
3412                                       xpath=NsrDtsHandler.SCALE_INSTANCE_XPATH,
3413                                       flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY| rwdts.Flag.CACHE,
3414                                       )
3415
3416             self._key_pair_regh = acg.register(
3417                                       xpath=NsrDtsHandler.KEY_PAIR_XPATH,
3418                                       flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY | rwdts.Flag.CACHE,
3419                                        )
3420
3421
3422 class NsrOpDataDtsHandler(object):
3423     """ The network service op data DTS handler """
3424     XPATH = "D,/nsr:ns-instance-opdata/nsr:nsr"
3425
3426     def __init__(self, dts, log, loop, nsm):
3427         self._dts = dts
3428         self._log = log
3429         self._loop = loop
3430         self._nsm = nsm
3431         self._regh = None
3432
3433     @property
3434     def regh(self):
3435         """ Return the registration handle"""
3436         return self._regh
3437
3438     @property
3439     def nsm(self):
3440         """ Return the NS manager instance """
3441         return self._nsm
3442
3443     @asyncio.coroutine
3444     def register(self):
3445         """ Register for Nsr op data publisher registration"""
3446         self._log.debug("Registering Nsr op data path %s as publisher",
3447                         NsrOpDataDtsHandler.XPATH)
3448
3449         hdl = rift.tasklets.DTS.RegistrationHandler()
3450         handlers = rift.tasklets.Group.Handler()
3451         with self._dts.group_create(handler=handlers) as group:
3452             self._regh = group.register(xpath=NsrOpDataDtsHandler.XPATH,
3453                                         handler=hdl,
3454                                         flags=rwdts.Flag.PUBLISHER | rwdts.Flag.NO_PREP_READ | rwdts.Flag.DATASTORE)
3455
3456     @asyncio.coroutine
3457     def create(self, path, msg):
3458         """
3459         Create an NS record in DTS with the path and message
3460         """
3461         self._log.debug("Creating NSR %s:%s", path, msg)
3462         self.regh.create_element(path, msg)
3463         self._log.debug("Created NSR, %s:%s", path, msg)
3464
3465     @asyncio.coroutine
3466     def update(self, path, msg, flags=rwdts.XactFlag.REPLACE):
3467         """
3468         Update an NS record in DTS with the path and message
3469         """
3470         self._log.debug("Updating NSR, %s:%s regh = %s", path, msg, self.regh)
3471         self.regh.update_element(path, msg, flags)
3472         self._log.debug("Updated NSR, %s:%s", path, msg)
3473
3474     @asyncio.coroutine
3475     def delete(self, path):
3476         """
3477         Update an NS record in DTS with the path and message
3478         """
3479         self._log.debug("Deleting NSR path:%s", path)
3480         self.regh.delete_element(path)
3481         self._log.debug("Deleted NSR path:%s", path)
3482
3483
3484 class VnfrDtsHandler(object):
3485     """ The virtual network service DTS handler """
3486     XPATH = "D,/vnfr:vnfr-catalog/vnfr:vnfr"
3487
3488     def __init__(self, dts, log, loop, nsm):
3489         self._dts = dts
3490         self._log = log
3491         self._loop = loop
3492         self._nsm = nsm
3493
3494         self._regh = None
3495
3496     @property
3497     def regh(self):
3498         """ Return registration handle """
3499         return self._regh
3500
3501     @property
3502     def nsm(self):
3503         """ Return the NS manager instance """
3504         return self._nsm
3505
3506     @asyncio.coroutine
3507     def register(self):
3508         """ Register for vnfr create/update/delete/ advises from dts """
3509
3510         def on_commit(xact_info):
3511             """ The transaction has been committed """
3512             self._log.debug("Got vnfr commit (xact_info: %s)", xact_info)
3513             return rwdts.MemberRspCode.ACTION_OK
3514
3515         @asyncio.coroutine
3516         def on_prepare(xact_info, action, ks_path, msg):
3517             """ prepare callback from dts """
3518             xpath = ks_path.to_xpath(RwNsrYang.get_schema())
3519             self._log.debug(
3520                 "Got vnfr on_prepare cb (xact_info: %s, action: %s): %s:%s",
3521                 xact_info, action, ks_path, msg
3522                 )
3523
3524             schema = VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr.schema()
3525             path_entry = schema.keyspec_to_entry(ks_path)
3526             if path_entry.key00.id not in self._nsm._vnfrs:
3527                 self._log.error("%s request for non existent record path %s",
3528                                 action, xpath)
3529                 xact_info.respond_xpath(rwdts.XactRspCode.NA, xpath)
3530
3531                 return
3532
3533                 self._log.debug("Deleting VNFR with id %s", path_entry.key00.id)
3534             if action == rwdts.QueryAction.CREATE or action == rwdts.QueryAction.UPDATE:
3535                 yield from self._nsm.update_vnfr(msg)
3536             elif action == rwdts.QueryAction.DELETE:
3537                 self._log.debug("Deleting VNFR with id %s", path_entry.key00.id)
3538                 self._nsm.delete_vnfr(path_entry.key00.id)
3539
3540             xact_info.respond_xpath(rwdts.XactRspCode.ACK, xpath)
3541
3542         self._log.debug("Registering for VNFR using xpath: %s",
3543                         VnfrDtsHandler.XPATH,)
3544
3545         hdl = rift.tasklets.DTS.RegistrationHandler(on_commit=on_commit,
3546                                                     on_prepare=on_prepare,)
3547         with self._dts.group_create() as group:
3548             self._regh = group.register(xpath=VnfrDtsHandler.XPATH,
3549                                         handler=hdl,
3550                                         flags=(rwdts.Flag.SUBSCRIBER),)
3551
3552
3553 class NsdRefCountDtsHandler(object):
3554     """ The NSD Ref Count DTS handler """
3555     XPATH = "D,/nsr:ns-instance-opdata/rw-nsr:nsd-ref-count"
3556
3557     def __init__(self, dts, log, loop, nsm):
3558         self._dts = dts
3559         self._log = log
3560         self._loop = loop
3561         self._nsm = nsm
3562
3563         self._regh = None
3564
3565     @property
3566     def regh(self):
3567         """ Return registration handle """
3568         return self._regh
3569
3570     @property
3571     def nsm(self):
3572         """ Return the NS manager instance """
3573         return self._nsm
3574
3575     @asyncio.coroutine
3576     def register(self):
3577         """ Register for NSD ref count read from dts """
3578
3579         @asyncio.coroutine
3580         def on_prepare(xact_info, action, ks_path, msg):
3581             """ prepare callback from dts """
3582             xpath = ks_path.to_xpath(RwNsrYang.get_schema())
3583
3584             if action == rwdts.QueryAction.READ:
3585                 schema = RwNsrYang.YangData_Nsr_NsInstanceOpdata_NsdRefCount.schema()
3586                 path_entry = schema.keyspec_to_entry(ks_path)
3587                 nsd_list = yield from self._nsm.get_nsd_refcount(path_entry.key00.nsd_id_ref)
3588                 for xpath, msg in nsd_list:
3589                     xact_info.respond_xpath(rsp_code=rwdts.XactRspCode.MORE,
3590                                             xpath=xpath,
3591                                             msg=msg)
3592                 xact_info.respond_xpath(rwdts.XactRspCode.ACK)
3593             else:
3594                 raise NetworkServiceRecordError("Not supported operation %s" % action)
3595
3596         hdl = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_prepare,)
3597         with self._dts.group_create() as group:
3598             self._regh = group.register(xpath=NsdRefCountDtsHandler.XPATH,
3599                                         handler=hdl,
3600                                         flags=rwdts.Flag.PUBLISHER,)
3601
3602
3603 class NsManager(object):
3604     """ The Network Service Manager class"""
3605     def __init__(self, dts, log, loop,
3606                  nsr_handler, vnfr_handler, vlr_handler, ro_plugin_selector,
3607                  vnffgmgr, vnfd_pub_handler, cloud_account_handler):
3608         self._dts = dts
3609         self._log = log
3610         self._loop = loop
3611         self._nsr_handler = nsr_handler
3612         self._vnfr_pub_handler = vnfr_handler
3613         self._vlr_pub_handler = vlr_handler
3614         self._vnffgmgr = vnffgmgr
3615         self._vnfd_pub_handler = vnfd_pub_handler
3616         self._cloud_account_handler = cloud_account_handler
3617
3618         self._ro_plugin_selector = ro_plugin_selector
3619         self._ncclient = rift.mano.ncclient.NcClient(
3620               host="127.0.0.1",
3621               port=2022,
3622               username="admin",
3623               password="admin",
3624               loop=self._loop)
3625
3626         self._nsrs = {}
3627         self._nsds = {}
3628         self._vnfds = {}
3629         self._vnfrs = {}
3630
3631         self.cfgmgr_obj = conman.ROConfigManager(log, loop, dts, self)
3632
3633         # TODO: All these handlers should move to tasklet level.
3634         # Passing self is often an indication of bad design
3635         self._nsd_dts_handler = NsdDtsHandler(dts, log, loop, self)
3636         self._vnfd_dts_handler = VnfdDtsHandler(dts, log, loop, self)
3637         self._dts_handlers = [self._nsd_dts_handler,
3638                               VnfrDtsHandler(dts, log, loop, self),
3639                               NsdRefCountDtsHandler(dts, log, loop, self),
3640                               NsrDtsHandler(dts, log, loop, self),
3641                               ScalingRpcHandler(log, dts, loop, self.scale_rpc_callback),
3642                               NsrRpcDtsHandler(dts,log,loop,self),
3643                               self._vnfd_dts_handler,
3644                               self.cfgmgr_obj,
3645                               ]
3646
3647
3648     @property
3649     def log(self):
3650         """ Log handle """
3651         return self._log
3652
3653     @property
3654     def loop(self):
3655         """ Loop """
3656         return self._loop
3657
3658     @property
3659     def dts(self):
3660         """ DTS handle """
3661         return self._dts
3662
3663     @property
3664     def nsr_handler(self):
3665         """" NSR handler """
3666         return self._nsr_handler
3667
3668     @property
3669     def so_obj(self):
3670         """" So Obj handler """
3671         return self._so_obj
3672
3673     @property
3674     def nsrs(self):
3675         """ NSRs in this NSM"""
3676         return self._nsrs
3677
3678     @property
3679     def nsds(self):
3680         """ NSDs in this NSM"""
3681         return self._nsds
3682
3683     @property
3684     def vnfds(self):
3685         """ VNFDs in this NSM"""
3686         return self._vnfds
3687
3688     @property
3689     def vnfrs(self):
3690         """ VNFRs in this NSM"""
3691         return self._vnfrs
3692
3693     @property
3694     def nsr_pub_handler(self):
3695         """ NSR publication handler """
3696         return self._nsr_handler
3697
3698     @property
3699     def vnfr_pub_handler(self):
3700         """ VNFR publication handler """
3701         return self._vnfr_pub_handler
3702
3703     @property
3704     def vlr_pub_handler(self):
3705         """ VLR publication handler """
3706         return self._vlr_pub_handler
3707
3708     @property
3709     def vnfd_pub_handler(self):
3710         return self._vnfd_pub_handler
3711
3712     @asyncio.coroutine
3713     def register(self):
3714         """ Register all static DTS handlers """
3715         for dts_handle in self._dts_handlers:
3716             yield from dts_handle.register()
3717
3718
3719     def get_ns_by_nsr_id(self, nsr_id):
3720         """ get NSR by nsr id """
3721         if nsr_id not in self._nsrs:
3722             raise NetworkServiceRecordError("NSR id %s not found" % nsr_id)
3723
3724         return self._nsrs[nsr_id]
3725
3726     def scale_nsr_out(self, nsr_id, scale_group_name, instance_id, config_xact):
3727         self.log.debug("Scale out NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3728                        nsr_id,
3729                        scale_group_name,
3730                        instance_id
3731                        )
3732         nsr = self._nsrs[nsr_id]
3733         if nsr.state != NetworkServiceRecordState.RUNNING:
3734             raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3735
3736         self._loop.create_task(nsr.create_scale_group_instance(scale_group_name, instance_id, config_xact))
3737
3738     def scale_nsr_in(self, nsr_id, scale_group_name, instance_id):
3739         self.log.debug("Scale in NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3740                        nsr_id,
3741                        scale_group_name,
3742                        instance_id,
3743                        )
3744         nsr = self._nsrs[nsr_id]
3745         if nsr.state != NetworkServiceRecordState.RUNNING:
3746             raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3747
3748         self._loop.create_task(nsr.delete_scale_group_instance(scale_group_name, instance_id))
3749
3750     def scale_rpc_callback(self, xact, msg, action):
3751         """Callback handler for RPC calls
3752         Args:
3753             xact : Transaction Handler
3754             msg : RPC input
3755             action : Scaling Action
3756         """
3757         ScalingGroupInstance = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup_Instance
3758         ScalingGroup = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
3759
3760         xpath = ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').format(
3761                           msg.nsr_id_ref)
3762         instance = ScalingGroupInstance.from_dict({"id": msg.instance_id})
3763
3764         @asyncio.coroutine
3765         def get_nsr_scaling_group():
3766             results = yield from self._dts.query_read(xpath, rwdts.XactFlag.MERGE)
3767
3768             for result in results:
3769                 res = yield from result
3770                 nsr_config = res.result
3771
3772             for scaling_group in nsr_config.scaling_group:
3773                 if scaling_group.scaling_group_name_ref == msg.scaling_group_name_ref:
3774                     break
3775             else:
3776                 scaling_group = nsr_config.scaling_group.add()
3777                 scaling_group.scaling_group_name_ref = msg.scaling_group_name_ref
3778
3779             return (nsr_config, scaling_group)
3780
3781         @asyncio.coroutine
3782         def update_config(nsr_config):
3783             xml = self._ncclient.convert_to_xml(RwNsrYang, nsr_config)
3784             xml = '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml)
3785             yield from self._ncclient.connect()
3786             yield from self._ncclient.manager.edit_config(target="running", config=xml, default_operation="replace")
3787
3788         @asyncio.coroutine
3789         def scale_out():
3790             nsr_config, scaling_group = yield from get_nsr_scaling_group()
3791             scaling_group.instance.append(instance)
3792             yield from update_config(nsr_config)
3793
3794         @asyncio.coroutine
3795         def scale_in():
3796             nsr_config, scaling_group = yield from get_nsr_scaling_group()
3797             scaling_group.instance.remove(instance)
3798             yield from update_config(nsr_config)
3799
3800         if action == ScalingRpcHandler.ACTION.SCALE_OUT:
3801             self._loop.create_task(scale_out())
3802         else:
3803             self._loop.create_task(scale_in())
3804
3805         # Opdata based calls, disabled for now!
3806         # if action == ScalingRpcHandler.ACTION.SCALE_OUT:
3807         #     self.scale_nsr_out(
3808         #           msg.nsr_id_ref,
3809         #           msg.scaling_group_name_ref,
3810         #           msg.instance_id,
3811         #           xact)
3812         # else:
3813         #     self.scale_nsr_in(
3814         #           msg.nsr_id_ref,
3815         #           msg.scaling_group_name_ref,
3816         #           msg.instance_id)
3817
3818     def nsr_update_cfg(self, nsr_id, msg):
3819         nsr = self._nsrs[nsr_id]
3820         nsr.nsr_cfg_msg= msg
3821
3822     def nsr_instantiate_vl(self, nsr_id, vld):
3823         self.log.debug("NSR {} create VL {}".format(nsr_id, vld))
3824         nsr = self._nsrs[nsr_id]
3825         if nsr.state != NetworkServiceRecordState.RUNNING:
3826             raise NsrVlUpdateError("Cannot perform VL instantiate if NSR is not in running state")
3827
3828         # Not calling in a separate task as this is called from a separate task
3829         yield from nsr.create_vl_instance(vld)
3830
3831     def nsr_terminate_vl(self, nsr_id, vld):
3832         self.log.debug("NSR {} delete VL {}".format(nsr_id, vld.id))
3833         nsr = self._nsrs[nsr_id]
3834         if nsr.state != NetworkServiceRecordState.RUNNING:
3835             raise NsrVlUpdateError("Cannot perform VL terminate if NSR is not in running state")
3836
3837         # Not calling in a separate task as this is called from a separate task
3838         yield from nsr.delete_vl_instance(vld)
3839
3840     def create_nsr(self, nsr_msg, key_pairs=None,restart_mode=False):
3841         """ Create an NSR instance """
3842         if nsr_msg.id in self._nsrs:
3843             msg = "NSR id %s already exists" % nsr_msg.id
3844             self._log.error(msg)
3845             raise NetworkServiceRecordError(msg)
3846
3847         self._log.info("Create NetworkServiceRecord nsr id %s from nsd_id %s",
3848                        nsr_msg.id,
3849                        nsr_msg.nsd.id)
3850
3851         nsm_plugin = self._ro_plugin_selector.ro_plugin
3852         sdn_account_name = self._cloud_account_handler.get_cloud_account_sdn_name(nsr_msg.cloud_account)
3853
3854         nsr = NetworkServiceRecord(self._dts,
3855                                    self._log,
3856                                    self._loop,
3857                                    self,
3858                                    nsm_plugin,
3859                                    nsr_msg,
3860                                    sdn_account_name,
3861                                    key_pairs,
3862                                    restart_mode=restart_mode,
3863                                    vlr_handler=self._ro_plugin_selector._records_publisher._vlr_pub_hdlr
3864                                    )
3865         self._nsrs[nsr_msg.id] = nsr
3866         nsm_plugin.create_nsr(nsr_msg, nsr_msg.nsd, key_pairs)
3867
3868         return nsr
3869
3870     def delete_nsr(self, nsr_id):
3871         """
3872         Delete NSR with the passed nsr id
3873         """
3874         del self._nsrs[nsr_id]
3875
3876     @asyncio.coroutine
3877     def instantiate_ns(self, nsr_id, config_xact):
3878         """ Instantiate an NS instance """
3879         self._log.debug("Instantiating Network service id %s", nsr_id)
3880         if nsr_id not in self._nsrs:
3881             err = "NSR id %s not found " % nsr_id
3882             self._log.error(err)
3883             raise NetworkServiceRecordError(err)
3884
3885         nsr = self._nsrs[nsr_id]
3886         yield from nsr.nsm_plugin.instantiate_ns(nsr, config_xact)
3887
3888     @asyncio.coroutine
3889     def update_vnfr(self, vnfr):
3890         """Create/Update an VNFR """
3891
3892         vnfr_state = self._vnfrs[vnfr.id].state
3893         self._log.debug("Updating VNFR with state %s: vnfr %s", vnfr_state, vnfr)
3894
3895         yield from self._vnfrs[vnfr.id].update_state(vnfr)
3896         nsr = self.find_nsr_for_vnfr(vnfr.id)
3897         yield from nsr.update_state()
3898
3899     def find_nsr_for_vnfr(self, vnfr_id):
3900         """ Find the NSR which )has the passed vnfr id"""
3901         for nsr in list(self.nsrs.values()):
3902             for vnfr in list(nsr.vnfrs.values()):
3903                 if vnfr.id == vnfr_id:
3904                     return nsr
3905         return None
3906
3907     def delete_vnfr(self, vnfr_id):
3908         """ Delete VNFR  with the passed id"""
3909         del self._vnfrs[vnfr_id]
3910
3911     def get_nsd_ref(self, nsd_id):
3912         """ Get network service descriptor for the passed nsd_id
3913             with a reference"""
3914         nsd = self.get_nsd(nsd_id)
3915         nsd.ref()
3916         return nsd
3917
3918     @asyncio.coroutine
3919     def get_nsr_config(self, nsd_id):
3920         xpath = "C,/nsr:ns-instance-config"
3921         results = yield from self._dts.query_read(xpath, rwdts.XactFlag.MERGE)
3922
3923         for result in results:
3924             entry = yield from result
3925             ns_instance_config = entry.result
3926
3927             for nsr in ns_instance_config.nsr:
3928                 if nsr.nsd.id == nsd_id:
3929                     return nsr
3930
3931         return None
3932
3933     @asyncio.coroutine
3934     def nsd_unref_by_nsr_id(self, nsr_id):
3935         """ Unref the network service descriptor based on NSR id """
3936         self._log.debug("NSR Unref called for Nsr Id:%s", nsr_id)
3937         if nsr_id in self._nsrs:
3938             nsr = self._nsrs[nsr_id]
3939
3940             try:
3941                 nsd = self.get_nsd(nsr.nsd_id)
3942                 self._log.debug("Releasing ref on NSD %s held by NSR %s - Curr %d",
3943                                 nsd.id, nsr.id, nsd.ref_count)
3944                 nsd.unref()
3945             except NetworkServiceDescriptorError:
3946                 # We store a copy of NSD in NSR and the NSD in nsd-catalog
3947                 # could be deleted
3948                 pass
3949
3950         else:
3951             self._log.error("Cannot find NSR with id %s", nsr_id)
3952             raise NetworkServiceDescriptorUnrefError("No NSR with id" % nsr_id)
3953
3954     @asyncio.coroutine
3955     def nsd_unref(self, nsd_id):
3956         """ Unref the network service descriptor associated with the id """
3957         nsd = self.get_nsd(nsd_id)
3958         nsd.unref()
3959
3960     def get_nsd(self, nsd_id):
3961         """ Get network service descriptor for the passed nsd_id"""
3962         if nsd_id not in self._nsds:
3963             self._log.error("Cannot find NSD id:%s", nsd_id)
3964             raise NetworkServiceDescriptorError("Cannot find NSD id:%s", nsd_id)
3965
3966         return self._nsds[nsd_id]
3967
3968     def create_nsd(self, nsd_msg):
3969         """ Create a network service descriptor """
3970         self._log.debug("Create network service descriptor - %s", nsd_msg)
3971         if nsd_msg.id in self._nsds:
3972             self._log.error("Cannot create NSD %s -NSD ID already exists", nsd_msg)
3973             raise NetworkServiceDescriptorError("NSD already exists-%s", nsd_msg.id)
3974
3975         nsd = NetworkServiceDescriptor(
3976                 self._dts,
3977                 self._log,
3978                 self._loop,
3979                 nsd_msg,
3980                 self
3981                 )
3982         self._nsds[nsd_msg.id] = nsd
3983
3984         return nsd
3985
3986     def update_nsd(self, nsd):
3987         """ update the Network service descriptor """
3988         self._log.debug("Update network service descriptor - %s", nsd)
3989         if nsd.id not in self._nsds:
3990             self._log.debug("No NSD found - creating NSD id = %s", nsd.id)
3991             self.create_nsd(nsd)
3992         else:
3993             self._log.debug("Updating NSD id = %s, nsd = %s", nsd.id, nsd)
3994             self._nsds[nsd.id].update(nsd)
3995
3996     def delete_nsd(self, nsd_id):
3997         """ Delete the Network service descriptor with the passed id """
3998         self._log.debug("Deleting the network service descriptor - %s", nsd_id)
3999         if nsd_id not in self._nsds:
4000             self._log.debug("Delete NSD failed - cannot find nsd-id %s", nsd_id)
4001             raise NetworkServiceDescriptorNotFound("Cannot find %s", nsd_id)
4002
4003         if nsd_id not in self._nsds:
4004             self._log.debug("Cannot delete NSD id %s reference exists %s",
4005                             nsd_id,
4006                             self._nsds[nsd_id].ref_count)
4007             raise NetworkServiceDescriptorRefCountExists(
4008                 "Cannot delete :%s, ref_count:%s",
4009                 nsd_id,
4010                 self._nsds[nsd_id].ref_count)
4011
4012         del self._nsds[nsd_id]
4013
4014     def get_vnfd_config(self, xact):
4015         vnfd_dts_reg = self._vnfd_dts_handler.regh
4016         for cfg in vnfd_dts_reg.get_xact_elements(xact):
4017             if cfg.id not in self._vnfds:
4018                 self.create_vnfd(cfg)
4019
4020     def get_vnfd(self, vnfd_id, xact):
4021         """ Get virtual network function descriptor for the passed vnfd_id"""
4022         if vnfd_id not in self._vnfds:
4023             self._log.error("Cannot find VNFD id:%s", vnfd_id)
4024             self.get_vnfd_config(xact)
4025
4026             if vnfd_id not in self._vnfds:
4027                 self._log.error("Cannot find VNFD id:%s", vnfd_id)
4028                 raise VnfDescriptorError("Cannot find VNFD id:%s", vnfd_id)
4029
4030         return self._vnfds[vnfd_id]
4031
4032     def create_vnfd(self, vnfd):
4033         """ Create a virtual network function descriptor """
4034         self._log.debug("Create virtual network function descriptor - %s", vnfd)
4035         if vnfd.id in self._vnfds:
4036             self._log.error("Cannot create VNFD %s -VNFD ID already exists", vnfd)
4037             raise VnfDescriptorError("VNFD already exists-%s", vnfd.id)
4038
4039         self._vnfds[vnfd.id] = vnfd
4040         return self._vnfds[vnfd.id]
4041
4042     def update_vnfd(self, vnfd):
4043         """ Update the virtual network function descriptor """
4044         self._log.debug("Update virtual network function descriptor- %s", vnfd)
4045
4046
4047         if vnfd.id not in self._vnfds:
4048             self._log.debug("No VNFD found - creating VNFD id = %s", vnfd.id)
4049             self.create_vnfd(vnfd)
4050         else:
4051             self._log.debug("Updating VNFD id = %s, vnfd = %s", vnfd.id, vnfd)
4052             self._vnfds[vnfd.id] = vnfd
4053
4054     @asyncio.coroutine
4055     def delete_vnfd(self, vnfd_id):
4056         """ Delete the virtual network function descriptor with the passed id """
4057         self._log.debug("Deleting the virtual network function descriptor - %s", vnfd_id)
4058         if vnfd_id not in self._vnfds:
4059             self._log.debug("Delete VNFD failed - cannot find vnfd-id %s", vnfd_id)
4060             raise VnfDescriptorError("Cannot find %s", vnfd_id)
4061
4062         del self._vnfds[vnfd_id]
4063
4064     def nsd_in_use(self, nsd_id):
4065         """ Is the NSD with the passed id in use """
4066         self._log.debug("Is this NSD in use - msg:%s", nsd_id)
4067         if nsd_id in self._nsds:
4068             return self._nsds[nsd_id].in_use()
4069         return False
4070
4071     @asyncio.coroutine
4072     def publish_nsr(self, xact, path, msg):
4073         """ Publish a NSR """
4074         self._log.debug("Publish NSR with path %s, msg %s",
4075                         path, msg)
4076         yield from self.nsr_handler.update(xact, path, msg)
4077
4078     @asyncio.coroutine
4079     def unpublish_nsr(self, xact, path):
4080         """ Un Publish an NSR """
4081         self._log.debug("Publishing delete NSR with path %s", path)
4082         yield from self.nsr_handler.delete(path, xact)
4083
4084     def vnfr_is_ready(self, vnfr_id):
4085         """ VNFR with the id is ready """
4086         self._log.debug("VNFR id %s ready", vnfr_id)
4087         if vnfr_id not in self._vnfds:
4088             err = "Did not find VNFR ID with id %s" % vnfr_id
4089             self._log.critical("err")
4090             raise VirtualNetworkFunctionRecordError(err)
4091         self._vnfrs[vnfr_id].is_ready()
4092
4093     @asyncio.coroutine
4094     def get_nsd_refcount(self, nsd_id):
4095         """ Get the nsd_list from this NSM"""
4096
4097         def nsd_refcount_xpath(nsd_id):
4098             """ xpath for ref count entry """
4099             return (NsdRefCountDtsHandler.XPATH +
4100                     "[rw-nsr:nsd-id-ref = '{}']").format(nsd_id)
4101
4102         nsd_list = []
4103         if nsd_id is None or nsd_id == "":
4104             for nsd in self._nsds.values():
4105                 nsd_msg = RwNsrYang.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4106                 nsd_msg.nsd_id_ref = nsd.id
4107                 nsd_msg.instance_ref_count = nsd.ref_count
4108                 nsd_list.append((nsd_refcount_xpath(nsd.id), nsd_msg))
4109         elif nsd_id in self._nsds:
4110             nsd_msg = RwNsrYang.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4111             nsd_msg.nsd_id_ref = self._nsds[nsd_id].id
4112             nsd_msg.instance_ref_count = self._nsds[nsd_id].ref_count
4113             nsd_list.append((nsd_refcount_xpath(nsd_id), nsd_msg))
4114
4115         return nsd_list
4116
4117     @asyncio.coroutine
4118     def terminate_ns(self, nsr_id, xact):
4119         """
4120         Terminate network service for the given NSR Id
4121         """
4122
4123         # Terminate the instances/networks assocaited with this nw service
4124         self._log.debug("Terminating the network service %s", nsr_id)
4125         try :
4126             yield from self._nsrs[nsr_id].terminate()
4127         except Exception as e:
4128             self.log.exception("Failed to terminate NSR[id=%s]", nsr_id)
4129
4130         # Unref the NSD
4131         yield from self.nsd_unref_by_nsr_id(nsr_id)
4132
4133         # Unpublish the NSR record
4134         self._log.debug("Unpublishing the network service %s", nsr_id)
4135         yield from self._nsrs[nsr_id].unpublish(xact)
4136
4137         # Finaly delete the NS instance from this NS Manager
4138         self._log.debug("Deletng the network service %s", nsr_id)
4139         self.delete_nsr(nsr_id)
4140
4141
4142 class NsmRecordsPublisherProxy(object):
4143     """ This class provides a publisher interface that allows plugin objects
4144         to publish NSR/VNFR/VLR"""
4145
4146     def __init__(self, dts, log, loop, nsr_pub_hdlr, vnfr_pub_hdlr, vlr_pub_hdlr):
4147         self._dts = dts
4148         self._log = log
4149         self._loop = loop
4150         self._nsr_pub_hdlr = nsr_pub_hdlr
4151         self._vlr_pub_hdlr = vlr_pub_hdlr
4152         self._vnfr_pub_hdlr = vnfr_pub_hdlr
4153
4154     @asyncio.coroutine
4155     def publish_nsr(self, xact, nsr):
4156         """ Publish an NSR """
4157         path = NetworkServiceRecord.xpath_from_nsr(nsr)
4158         return (yield from self._nsr_pub_hdlr.update(xact, path, nsr))
4159
4160     @asyncio.coroutine
4161     def unpublish_nsr(self, xact, nsr):
4162         """ Unpublish an NSR """
4163         path = NetworkServiceRecord.xpath_from_nsr(nsr)
4164         return (yield from self._nsr_pub_hdlr.delete(xact, path))
4165
4166     @asyncio.coroutine
4167     def publish_vnfr(self, xact, vnfr):
4168         """ Publish an VNFR """
4169         path = VirtualNetworkFunctionRecord.vnfr_xpath(vnfr)
4170         return (yield from self._vnfr_pub_hdlr.update(xact, path, vnfr))
4171
4172     @asyncio.coroutine
4173     def unpublish_vnfr(self, xact, vnfr):
4174         """ Unpublish a VNFR """
4175         path = VirtualNetworkFunctionRecord.vnfr_xpath(vnfr)
4176         return (yield from self._vnfr_pub_hdlr.delete(xact, path))
4177
4178     @asyncio.coroutine
4179     def publish_vlr(self, xact, vlr):
4180         """ Publish a VLR """
4181         path = VirtualLinkRecord.vlr_xpath(vlr)
4182         return (yield from self._vlr_pub_hdlr.update(xact, path, vlr))
4183
4184     @asyncio.coroutine
4185     def unpublish_vlr(self, xact, vlr):
4186         """ Unpublish a VLR """
4187         path = VirtualLinkRecord.vlr_xpath(vlr)
4188         return (yield from self._vlr_pub_hdlr.delete(xact, path))
4189
4190
4191 class ScalingRpcHandler(mano_dts.DtsHandler):
4192     """ The Network service Monitor DTS handler """
4193     SCALE_IN_INPUT_XPATH = "I,/nsr:exec-scale-in"
4194     SCALE_IN_OUTPUT_XPATH = "O,/nsr:exec-scale-in"
4195
4196     SCALE_OUT_INPUT_XPATH = "I,/nsr:exec-scale-out"
4197     SCALE_OUT_OUTPUT_XPATH = "O,/nsr:exec-scale-out"
4198
4199     ACTION = Enum('ACTION', 'SCALE_IN SCALE_OUT')
4200
4201     def __init__(self, log, dts, loop, callback=None):
4202         super().__init__(log, dts, loop)
4203         self.callback = callback
4204         self.last_instance_id = defaultdict(int)
4205
4206     @asyncio.coroutine
4207     def register(self):
4208
4209         @asyncio.coroutine
4210         def on_scale_in_prepare(xact_info, action, ks_path, msg):
4211             assert action == rwdts.QueryAction.RPC
4212
4213             try:
4214                 if self.callback:
4215                     self.callback(xact_info.xact, msg, self.ACTION.SCALE_IN)
4216
4217                 rpc_op = NsrYang.YangOutput_Nsr_ExecScaleIn.from_dict({
4218                       "instance_id": msg.instance_id})
4219
4220                 xact_info.respond_xpath(
4221                     rwdts.XactRspCode.ACK,
4222                     self.__class__.SCALE_IN_OUTPUT_XPATH,
4223                     rpc_op)
4224
4225             except Exception as e:
4226                 self.log.exception(e)
4227                 xact_info.respond_xpath(
4228                     rwdts.XactRspCode.NACK,
4229                     self.__class__.SCALE_IN_OUTPUT_XPATH)
4230
4231         @asyncio.coroutine
4232         def on_scale_out_prepare(xact_info, action, ks_path, msg):
4233             assert action == rwdts.QueryAction.RPC
4234
4235             try:
4236                 scaling_group = msg.scaling_group_name_ref
4237                 if not msg.instance_id:
4238                     last_instance_id = self.last_instance_id[scale_group]
4239                     msg.instance_id  = last_instance_id + 1
4240                     self.last_instance_id[scale_group] += 1
4241
4242                 if self.callback:
4243                     self.callback(xact_info.xact, msg, self.ACTION.SCALE_OUT)
4244
4245                 rpc_op = NsrYang.YangOutput_Nsr_ExecScaleOut.from_dict({
4246                       "instance_id": msg.instance_id})
4247
4248                 xact_info.respond_xpath(
4249                     rwdts.XactRspCode.ACK,
4250                     self.__class__.SCALE_OUT_OUTPUT_XPATH,
4251                     rpc_op)
4252
4253             except Exception as e:
4254                 self.log.exception(e)
4255                 xact_info.respond_xpath(
4256                       rwdts.XactRspCode.NACK,
4257                       self.__class__.SCALE_OUT_OUTPUT_XPATH)
4258
4259         scale_in_hdl = rift.tasklets.DTS.RegistrationHandler(
4260               on_prepare=on_scale_in_prepare)
4261         scale_out_hdl = rift.tasklets.DTS.RegistrationHandler(
4262               on_prepare=on_scale_out_prepare)
4263
4264         with self.dts.group_create() as group:
4265             group.register(
4266                   xpath=self.__class__.SCALE_IN_INPUT_XPATH,
4267                   handler=scale_in_hdl,
4268                   flags=rwdts.Flag.PUBLISHER)
4269             group.register(
4270                   xpath=self.__class__.SCALE_OUT_INPUT_XPATH,
4271                   handler=scale_out_hdl,
4272                   flags=rwdts.Flag.PUBLISHER)
4273
4274
4275 class NsmTasklet(rift.tasklets.Tasklet):
4276     """
4277     The network service manager  tasklet
4278     """
4279     def __init__(self, *args, **kwargs):
4280         super(NsmTasklet, self).__init__(*args, **kwargs)
4281         self.rwlog.set_category("rw-mano-log")
4282         self.rwlog.set_subcategory("nsm")
4283
4284         self._dts = None
4285         self._nsm = None
4286
4287         self._ro_plugin_selector = None
4288         self._vnffgmgr = None
4289
4290         self._nsr_handler = None
4291         self._vnfr_pub_handler = None
4292         self._vlr_pub_handler = None
4293         self._vnfd_pub_handler = None
4294         self._scale_cfg_handler = None
4295
4296         self._records_publisher_proxy = None
4297
4298     def start(self):
4299         """ The task start callback """
4300         super(NsmTasklet, self).start()
4301         self.log.info("Starting NsmTasklet")
4302
4303         self.log.debug("Registering with dts")
4304         self._dts = rift.tasklets.DTS(self.tasklet_info,
4305                                       RwNsmYang.get_schema(),
4306                                       self.loop,
4307                                       self.on_dts_state_change)
4308
4309         self.log.debug("Created DTS Api GI Object: %s", self._dts)
4310
4311     def stop(self):
4312         try:
4313             self._dts.deinit()
4314         except Exception:
4315             print("Caught Exception in NSM stop:", sys.exc_info()[0])
4316             raise
4317
4318     def on_instance_started(self):
4319         """ Task instance started callback """
4320         self.log.debug("Got instance started callback")
4321
4322     @asyncio.coroutine
4323     def init(self):
4324         """ Task init callback """
4325         self.log.debug("Got instance started callback")
4326
4327         self.log.debug("creating config account handler")
4328
4329         self._nsr_pub_handler = publisher.NsrOpDataDtsHandler(self._dts, self.log, self.loop)
4330         yield from self._nsr_pub_handler.register()
4331
4332         self._vnfr_pub_handler = publisher.VnfrPublisherDtsHandler(self._dts, self.log, self.loop)
4333         yield from self._vnfr_pub_handler.register()
4334
4335         self._vlr_pub_handler = publisher.VlrPublisherDtsHandler(self._dts, self.log, self.loop)
4336         yield from self._vlr_pub_handler.register()
4337
4338         manifest = self.tasklet_info.get_pb_manifest()
4339         use_ssl = manifest.bootstrap_phase.rwsecurity.use_ssl
4340         ssl_cert = manifest.bootstrap_phase.rwsecurity.cert
4341         ssl_key = manifest.bootstrap_phase.rwsecurity.key
4342
4343         self._vnfd_pub_handler = publisher.VnfdPublisher(use_ssl, ssl_cert, ssl_key, self.loop)
4344
4345         self._records_publisher_proxy = NsmRecordsPublisherProxy(
4346                 self._dts,
4347                 self.log,
4348                 self.loop,
4349                 self._nsr_pub_handler,
4350                 self._vnfr_pub_handler,
4351                 self._vlr_pub_handler,
4352                 )
4353
4354         # Register the NSM to receive the nsm plugin
4355         # when cloud account is configured
4356         self._ro_plugin_selector = cloud.ROAccountPluginSelector(
4357                 self._dts,
4358                 self.log,
4359                 self.loop,
4360                 self._records_publisher_proxy,
4361                 )
4362         yield from self._ro_plugin_selector.register()
4363
4364         self._cloud_account_handler = cloud.CloudAccountConfigSubscriber(
4365                 self._log,
4366                 self._dts,
4367                 self.log_hdl)
4368
4369         yield from self._cloud_account_handler.register()
4370
4371         self._vnffgmgr = rwvnffgmgr.VnffgMgr(self._dts, self.log, self.log_hdl, self.loop)
4372         yield from self._vnffgmgr.register()
4373
4374         self._nsm = NsManager(
4375                 self._dts,
4376                 self.log,
4377                 self.loop,
4378                 self._nsr_pub_handler,
4379                 self._vnfr_pub_handler,
4380                 self._vlr_pub_handler,
4381                 self._ro_plugin_selector,
4382                 self._vnffgmgr,
4383                 self._vnfd_pub_handler,
4384                 self._cloud_account_handler
4385                 )
4386
4387         yield from self._nsm.register()
4388
4389     @asyncio.coroutine
4390     def run(self):
4391         """ Task run callback """
4392         pass
4393
4394     @asyncio.coroutine
4395     def on_dts_state_change(self, state):
4396         """Take action according to current dts state to transition
4397         application into the corresponding application state
4398
4399         Arguments
4400             state - current dts state
4401         """
4402         switch = {
4403             rwdts.State.INIT: rwdts.State.REGN_COMPLETE,
4404             rwdts.State.CONFIG: rwdts.State.RUN,
4405         }
4406
4407         handlers = {
4408             rwdts.State.INIT: self.init,
4409             rwdts.State.RUN: self.run,
4410         }
4411
4412         # Transition application to next state
4413         handler = handlers.get(state, None)
4414         if handler is not None:
4415             yield from handler()
4416
4417         # Transition dts to next state
4418         next_state = switch.get(state, None)
4419         if next_state is not None:
4420             self.log.debug("Changing state to %s", next_state)
4421             self._dts.handle.set_state(next_state)