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

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