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