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