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