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     RwsdnYang,
  55     RwDts as rwdts,
  56     RwTypes,
  57     ProtobufC,
  58 )
  59
  60 import rift.tasklets
  61 import rift.mano.ncclient
  62 import rift.mano.config_data.config
  63 import rift.mano.dts as mano_dts
  64 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 =  RwsdnYang.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.error("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 __call__(self, nsd, nsr_config):
2686         """Substitutes input parameters from the NSR config into the NSD
2687
2688         This call modifies the provided NSD with the input parameters that are
2689         contained in the NSR config.
2690
2691         Arguments:
2692             nsd        - a GI NSD object
2693             nsr_config - a GI NSR config object
2694
2695         """
2696         if nsd is None or nsr_config is None:
2697             return
2698
2699         # Create a lookup of the xpath elements that this descriptor allows
2700         # to be modified
2701         optional_input_parameters = set()
2702         for input_parameter in nsd.input_parameter_xpath:
2703             optional_input_parameters.add(self.project.add_project(input_parameter.xpath))
2704
2705         # Apply the input parameters to the descriptor
2706         if nsr_config.input_parameter:
2707             for param in nsr_config.input_parameter:
2708                 if param.xpath not in optional_input_parameters:
2709                     msg = "tried to set an invalid input parameter ({})"
2710                     self.log.error(msg.format(param.xpath))
2711                     continue
2712
2713                 self.log.debug(
2714                         "input-parameter:{} = {}".format(
2715                             param.xpath,
2716                             param.value,
2717                             )
2718                         )
2719
2720                 try:
2721                     xpath.setxattr(nsd, param.xpath, param.value)
2722
2723                 except Exception as e:
2724                     self.log.exception(e)
2725
2726
2727 class NetworkServiceDescriptor(object):
2728     """
2729     Network service descriptor class
2730     """
2731
2732     def __init__(self, dts, log, loop, nsd, nsm):
2733         self._dts = dts
2734         self._log = log
2735         self._loop = loop
2736
2737         self._nsd = nsd
2738         self._ref_count = 0
2739
2740         self._nsm = nsm
2741
2742     @property
2743     def id(self):
2744         """ Returns nsd id """
2745         return self._nsd.id
2746
2747     @property
2748     def name(self):
2749         """ Returns name of nsd """
2750         return self._nsd.name
2751
2752     @property
2753     def ref_count(self):
2754         """ Returns reference count"""
2755         return self._ref_count
2756
2757     def in_use(self):
2758         """ Returns whether nsd is in use or not """
2759         return True if self.ref_count > 0 else False
2760
2761     def ref(self):
2762         """ Take a reference on this object """
2763         self._ref_count += 1
2764
2765     def unref(self):
2766         """ Release reference on this object """
2767         if self.ref_count < 1:
2768             msg = ("Unref on a NSD object - nsd id %s, ref_count = %s" %
2769                    (self.id, self.ref_count))
2770             self._log.critical(msg)
2771             raise NetworkServiceDescriptorError(msg)
2772         self._ref_count -= 1
2773
2774     @property
2775     def msg(self):
2776         """ Return the message associated with this NetworkServiceDescriptor"""
2777         return self._nsd
2778
2779     @staticmethod
2780     def path_for_id(nsd_id):
2781         """ Return path for the passed nsd_id"""
2782         return self._nsm._project.add_project(
2783             "C,/project-nsd:nsd-catalog/project-nsd:nsd[project-nsd:id = '{}'".
2784             format(nsd_id))
2785
2786     def path(self):
2787         """ Return the message associated with this NetworkServiceDescriptor"""
2788         return NetworkServiceDescriptor.path_for_id(self.id)
2789
2790     def update(self, nsd):
2791         """ Update the NSD descriptor """
2792         self._nsd = nsd
2793
2794
2795 class NsdDtsHandler(object):
2796     """ The network service descriptor DTS handler """
2797     XPATH = "C,/project-nsd:nsd-catalog/project-nsd:nsd"
2798
2799     def __init__(self, dts, log, loop, nsm):
2800         self._dts = dts
2801         self._log = log
2802         self._loop = loop
2803         self._nsm = nsm
2804
2805         self._regh = None
2806         self._project = nsm._project
2807
2808     @property
2809     def regh(self):
2810         """ Return registration handle """
2811         return self._regh
2812
2813     @asyncio.coroutine
2814     def register(self):
2815         """ Register for Nsd create/update/delete/read requests from dts """
2816
2817         if self._regh:
2818             self._log.warning("DTS handler already registered for project {}".
2819                               format(self._project.name))
2820             return
2821
2822         def on_apply(dts, acg, xact, action, scratch):
2823             """Apply the  configuration"""
2824             is_recovery = xact.xact is None and action == rwdts.AppconfAction.INSTALL
2825             self._log.debug("Got nsd apply cfg (xact:%s) (action:%s)",
2826                             xact, action)
2827
2828             if self._regh:
2829                 # Create/Update an NSD record
2830                 for cfg in self._regh.get_xact_elements(xact):
2831                     # Only interested in those NSD cfgs whose ID was received in prepare callback
2832                     if cfg.id in scratch.get('nsds', []) or is_recovery:
2833                         self._nsm.update_nsd(cfg)
2834
2835             else:
2836                 self._log.error("No reg handle for {} for project {}".
2837                                 format(self.__class__, self._project.name))
2838
2839             scratch.pop('nsds', None)
2840
2841             return RwTypes.RwStatus.SUCCESS
2842
2843         @asyncio.coroutine
2844         def delete_nsd_libs(nsd_id):
2845             """ Remove any files uploaded with NSD and stored under $RIFT_ARTIFACTS/libs/<id> """
2846             try:
2847                 rift_artifacts_dir = os.environ['RIFT_ARTIFACTS']
2848                 nsd_dir = os.path.join(rift_artifacts_dir, 'launchpad/libs', nsd_id)
2849
2850                 if os.path.exists (nsd_dir):
2851                     shutil.rmtree(nsd_dir, ignore_errors=True)
2852             except Exception as e:
2853                 self._log.error("Exception in cleaning up NSD libs {}: {}".
2854                                 format(nsd_id, e))
2855                 self._log.exception(e)
2856
2857         @asyncio.coroutine
2858         def on_prepare(dts, acg, xact, xact_info, ks_path, msg, scratch):
2859             """ Prepare callback from DTS for NSD config """
2860
2861             self._log.info("Got nsd prepare - config received nsd id %s, msg %s",
2862                            msg.id, msg)
2863
2864             fref = ProtobufC.FieldReference.alloc()
2865             fref.goto_whole_message(msg.to_pbcm())
2866
2867             if fref.is_field_deleted():
2868                 # Delete an NSD record
2869                 self._log.debug("Deleting NSD with id %s", msg.id)
2870                 if self._nsm.nsd_in_use(msg.id):
2871                     self._log.debug("Cannot delete NSD in use - %s", msg.id)
2872                     err = "Cannot delete an NSD in use - %s" % msg.id
2873                     raise NetworkServiceDescriptorRefCountExists(err)
2874
2875                 yield from delete_nsd_libs(msg.id)
2876                 self._nsm.delete_nsd(msg.id)
2877             else:
2878                 # Add this NSD to scratch to create/update in apply callback
2879                 nsds = scratch.setdefault('nsds', [])
2880                 nsds.append(msg.id)
2881                 # acg._scratch['nsds'].append(msg.id)
2882
2883             xact_info.respond_xpath(rwdts.XactRspCode.ACK)
2884
2885         self._log.debug(
2886             "Registering for NSD config using xpath: %s",
2887             NsdDtsHandler.XPATH,
2888             )
2889
2890         acg_hdl = rift.tasklets.AppConfGroup.Handler(on_apply=on_apply)
2891         with self._dts.appconf_group_create(handler=acg_hdl) as acg:
2892             # Need a list in scratch to store NSDs to create/update later
2893             # acg._scratch['nsds'] = list()
2894             self._regh = acg.register(
2895                 xpath=self._project.add_project(NsdDtsHandler.XPATH),
2896                 flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY | rwdts.Flag.CACHE,
2897                 on_prepare=on_prepare)
2898
2899     def deregister(self):
2900         self._log.debug("De-register NSD handler for project {}".
2901                         format(self._project.name))
2902         if self._regh:
2903             self._regh.deregister()
2904             self._regh = None
2905
2906
2907 class VnfdDtsHandler(object):
2908     """ DTS handler for VNFD config changes """
2909     XPATH = "C,/project-vnfd:vnfd-catalog/project-vnfd:vnfd"
2910
2911     def __init__(self, dts, log, loop, nsm):
2912         self._dts = dts
2913         self._log = log
2914         self._loop = loop
2915         self._nsm = nsm
2916         self._regh = None
2917         self._project = nsm._project
2918
2919     @property
2920     def regh(self):
2921         """ DTS registration handle """
2922         return self._regh
2923
2924     @asyncio.coroutine
2925     def register(self):
2926         """ Register for VNFD configuration"""
2927
2928         if self._regh:
2929             self._log.warning("DTS handler already registered for project {}".
2930                               format(self._project.name))
2931             return
2932
2933         @asyncio.coroutine
2934         def on_apply(dts, acg, xact, action, scratch):
2935             """Apply the  configuration"""
2936             self._log.debug("Got NSM VNFD apply (xact: %s) (action: %s)(scr: %s)",
2937                             xact, action, scratch)
2938
2939             if self._regh:
2940                 # Create/Update a VNFD record
2941                 for cfg in self._regh.get_xact_elements(xact):
2942                     # Only interested in those VNFD cfgs whose ID was received in prepare callback
2943                     if cfg.id in scratch.get('vnfds', []):
2944                         self._nsm.update_vnfd(cfg)
2945
2946                         for cfg in self._regh.elements:
2947                             if cfg.id in scratch.get('deleted_vnfds', []):
2948                                 yield from self._nsm.delete_vnfd(cfg.id)
2949
2950             else:
2951                 self._log.error("Reg handle none for {} in project {}".
2952                                 format(self.__class__, self._project))
2953
2954             scratch.pop('vnfds', None)
2955             scratch.pop('deleted_vnfds', None)
2956
2957         @asyncio.coroutine
2958         def on_prepare(dts, acg, xact, xact_info, ks_path, msg, scratch):
2959             """ on prepare callback """
2960             self._log.debug("Got on prepare for VNFD (path: %s) (action: %s) (msg: %s)",
2961                             ks_path.to_xpath(RwNsmYang.get_schema()), xact_info.query_action, msg)
2962
2963             fref = ProtobufC.FieldReference.alloc()
2964             fref.goto_whole_message(msg.to_pbcm())
2965
2966             # Handle deletes in prepare_callback, but adds/updates in apply_callback
2967             if fref.is_field_deleted():
2968                 self._log.debug("Adding msg to deleted field")
2969                 deleted_vnfds = scratch.setdefault('deleted_vnfds', [])
2970                 deleted_vnfds.append(msg.id)
2971             else:
2972                 # Add this VNFD to scratch to create/update in apply callback
2973                 vnfds = scratch.setdefault('vnfds', [])
2974                 vnfds.append(msg.id)
2975
2976             xact_info.respond_xpath(rwdts.XactRspCode.ACK)
2977
2978         xpath = self._project.add_project(VnfdDtsHandler.XPATH)
2979         self._log.debug(
2980             "Registering for VNFD config using xpath {} for project {}"
2981             .format(xpath, self._project))
2982         acg_hdl = rift.tasklets.AppConfGroup.Handler(on_apply=on_apply)
2983         with self._dts.appconf_group_create(handler=acg_hdl) as acg:
2984             # Need a list in scratch to store VNFDs to create/update later
2985             # acg._scratch['vnfds'] = list()
2986             # acg._scratch['deleted_vnfds'] = list()
2987             self._regh = acg.register(
2988                 xpath=xpath,
2989                 flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY,
2990                 on_prepare=on_prepare)
2991
2992     def deregister(self):
2993         self._log.debug("De-register VNFD handler for project {}".
2994                         format(self._project.name))
2995         if self._regh:
2996             self._regh.deregister()
2997             self._regh = None
2998
2999
3000 class NsrRpcDtsHandler(object):
3001     """ The network service instantiation RPC DTS handler """
3002     EXEC_NSR_CONF_XPATH = "I,/nsr:start-network-service"
3003     EXEC_NSR_CONF_O_XPATH = "O,/nsr:start-network-service"
3004     NETCONF_IP_ADDRESS = "127.0.0.1"
3005     NETCONF_PORT = 2022
3006     RESTCONF_PORT = 8888
3007     NETCONF_USER = "admin"
3008     NETCONF_PW = "admin"
3009     REST_BASE_V2_URL = 'https://{}:{}/v2/api/'.format("127.0.0.1",8888)
3010
3011     def __init__(self, dts, log, loop, nsm):
3012         self._dts = dts
3013         self._log = log
3014         self._loop = loop
3015         self._nsm = nsm
3016         self._nsd = None
3017
3018         self._ns_regh = None
3019
3020         self._manager = None
3021         self._nsr_config_url = NsrRpcDtsHandler.REST_BASE_V2_URL + \
3022                                'config/project/{}/ns-instance-config'. \
3023                                format(self._nsm._project.name)
3024
3025         self._model = RwYang.Model.create_libncx()
3026         self._model.load_schema_ypbc(RwNsrYang.get_schema())
3027
3028     @property
3029     def nsm(self):
3030         """ Return the NS manager instance """
3031         return self._nsm
3032
3033     @staticmethod
3034     def wrap_netconf_config_xml(xml):
3035         xml = '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml)
3036         return xml
3037
3038     @asyncio.coroutine
3039     def _connect(self, timeout_secs=240):
3040
3041         start_time = time.time()
3042         while (time.time() - start_time) < timeout_secs:
3043
3044             try:
3045                 self._log.debug("Attemping NsmTasklet netconf connection.")
3046
3047                 manager = yield from ncclient.asyncio_manager.asyncio_connect(
3048                         loop=self._loop,
3049                         host=NsrRpcDtsHandler.NETCONF_IP_ADDRESS,
3050                         port=NsrRpcDtsHandler.NETCONF_PORT,
3051                         username=NsrRpcDtsHandler.NETCONF_USER,
3052                         password=NsrRpcDtsHandler.NETCONF_PW,
3053                         allow_agent=False,
3054                         look_for_keys=False,
3055                         hostkey_verify=False,
3056                         )
3057
3058                 return manager
3059
3060             except ncclient.transport.errors.SSHError as e:
3061                 self._log.warning("Netconf connection to launchpad %s failed: %s",
3062                                   NsrRpcDtsHandler.NETCONF_IP_ADDRESS, str(e))
3063
3064             yield from asyncio.sleep(5, loop=self._loop)
3065
3066         raise NsrInstantiationFailed("Failed to connect to Launchpad within %s seconds" %
3067                                       timeout_secs)
3068
3069     def _apply_ns_instance_config(self,payload_dict):
3070         #self._log.debug("At apply NS instance config with payload %s",payload_dict)
3071         req_hdr= {'accept':'application/vnd.yang.data+json',
3072                   'content-type':'application/vnd.yang.data+json'}
3073         response=requests.post(self._nsr_config_url, headers=req_hdr,
3074                                auth=('admin', 'admin'),data=payload_dict,verify=False)
3075         return response
3076
3077     @asyncio.coroutine
3078     def register(self):
3079         """ Register for NS monitoring read from dts """
3080         if self._ns_regh:
3081             self._log.warning("RPC already registered for project {}".
3082                               format(self._project.name))
3083             return
3084
3085         @asyncio.coroutine
3086         def on_ns_config_prepare(xact_info, action, ks_path, msg):
3087             """ prepare callback from dts start-network-service"""
3088             assert action == rwdts.QueryAction.RPC
3089             rpc_ip = msg
3090
3091             if not self._nsm._project.rpc_check(msg, xact_info=xact_info):
3092                 return
3093
3094             rpc_op = NsrYang.YangOutput_Nsr_StartNetworkService.from_dict({
3095                 "nsr_id":str(uuid.uuid4()),
3096                 "project_name": msg.prject_name,
3097             })
3098
3099             if not ('name' in rpc_ip and  'nsd_ref' in rpc_ip and
3100                     ('cloud_account' in rpc_ip or 'om_datacenter' in rpc_ip)):
3101                 self._log.error("Mandatory parameters name or nsd_ref or " +
3102                                 "cloud account not found in start-network-service {}".
3103                                 format(rpc_ip))
3104
3105
3106             self._log.debug("start-network-service RPC input: {}".format(rpc_ip))
3107
3108             try:
3109                 # Add used value to the pool
3110                 self._log.debug("RPC output: {}".format(rpc_op))
3111
3112                 nsd_copy = self.nsm.get_nsd(rpc_ip.nsd_ref)
3113
3114                 #if not self._manager:
3115                 #    self._manager = yield from self._connect()
3116
3117                 self._log.debug("Configuring ns-instance-config with name  %s nsd-ref: %s",
3118                         rpc_ip.name, rpc_ip.nsd_ref)
3119
3120                 ns_instance_config_dict = {"id":rpc_op.nsr_id, "admin_status":"ENABLED"}
3121                 ns_instance_config_copy_dict = {k:v for k, v in rpc_ip.as_dict().items()
3122                                                 if k in RwNsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr().fields}
3123                 ns_instance_config_dict.update(ns_instance_config_copy_dict)
3124
3125                 ns_instance_config = RwNsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr.from_dict(ns_instance_config_dict)
3126                 ns_instance_config.nsd = NsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr_Nsd()
3127                 ns_instance_config.nsd.from_dict(nsd_copy.msg.as_dict())
3128
3129                 payload_dict = ns_instance_config.to_json(self._model)
3130                 #xml = ns_instance_config.to_xml_v2(self._model)
3131                 #netconf_xml = self.wrap_netconf_config_xml(xml)
3132
3133                 #self._log.debug("Sending configure ns-instance-config xml to %s: %s",
3134                 #        netconf_xml, NsrRpcDtsHandler.NETCONF_IP_ADDRESS)
3135                 self._log.debug("Sending configure ns-instance-config json to %s: %s",
3136                         self._nsr_config_url,ns_instance_config)
3137
3138                 #response = yield from self._manager.edit_config(
3139                 #           target="running",
3140                 #           config=netconf_xml,
3141                 #           )
3142                 response = yield from self._loop.run_in_executor(
3143                     None,
3144                     self._apply_ns_instance_config,
3145                     payload_dict
3146                     )
3147                 response.raise_for_status()
3148                 self._log.debug("Received edit config response: %s", response.json())
3149
3150                 xact_info.respond_xpath(rwdts.XactRspCode.ACK,
3151                                         NsrRpcDtsHandler.EXEC_NSR_CONF_O_XPATH,
3152                                         rpc_op)
3153             except Exception as e:
3154                 self._log.error("Exception processing the "
3155                                 "start-network-service: {}".format(e))
3156                 self._log.exception(e)
3157                 xact_info.respond_xpath(rwdts.XactRspCode.NACK,
3158                                         NsrRpcDtsHandler.EXEC_NSR_CONF_O_XPATH)
3159
3160
3161         hdl_ns = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_ns_config_prepare,)
3162
3163         with self._dts.group_create() as group:
3164             self._ns_regh = group.register(xpath=NsrRpcDtsHandler.EXEC_NSR_CONF_XPATH,
3165                                            handler=hdl_ns,
3166                                            flags=rwdts.Flag.PUBLISHER,
3167                                           )
3168
3169     def deregister(self):
3170         self._log.debug("De-register NSR RPC for project {}".
3171                         format(self._nsm._project.name))
3172         if self._ns_regh:
3173             self._ns_regh.deregister()
3174             self._ns_regh = None
3175
3176
3177 class NsrDtsHandler(object):
3178     """ The network service DTS handler """
3179     NSR_XPATH = "C,/nsr:ns-instance-config/nsr:nsr"
3180     SCALE_INSTANCE_XPATH = "C,/nsr:ns-instance-config/nsr:nsr/nsr:scaling-group/nsr:instance"
3181     KEY_PAIR_XPATH = "C,/nsr:key-pair"
3182
3183     def __init__(self, dts, log, loop, nsm):
3184         self._dts = dts
3185         self._log = log
3186         self._loop = loop
3187         self._nsm = nsm
3188         self._project = self._nsm._project
3189
3190         self._nsr_regh = None
3191         self._scale_regh = None
3192         self._key_pair_regh = None
3193
3194     @property
3195     def nsm(self):
3196         """ Return the NS manager instance """
3197         return self._nsm
3198
3199     @asyncio.coroutine
3200     def register(self):
3201         """ Register for Nsr create/update/delete/read requests from dts """
3202
3203         if self._nsr_regh:
3204             self._log.warning("DTS handler already registered for project {}".
3205                               format(self._project.name))
3206             return
3207
3208         def nsr_id_from_keyspec(ks):
3209             nsr_path_entry = NsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr.schema().keyspec_to_entry(ks)
3210             nsr_id = nsr_path_entry.key00.id
3211             return nsr_id
3212
3213         def group_name_from_keyspec(ks):
3214             group_path_entry = NsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup.schema().keyspec_to_entry(ks)
3215             group_name = group_path_entry.key00.scaling_group_name_ref
3216             return group_name
3217
3218         def is_instance_in_reg_elements(nsr_id, group_name, instance_id):
3219             """ Return boolean indicating if scaling group instance was already commited previously.
3220
3221             By looking at the existing elements in this registration handle (elements not part
3222             of this current xact), we can tell if the instance was configured previously without
3223             keeping any application state.
3224             """
3225             for instance_cfg, keyspec in self._nsr_regh.get_xact_elements(include_keyspec=True):
3226                 elem_nsr_id = nsr_id_from_keyspec(keyspec)
3227                 elem_group_name = group_name_from_keyspec(keyspec)
3228
3229                 if elem_nsr_id != nsr_id or group_name != elem_group_name:
3230                     continue
3231
3232                 if instance_cfg.id == instance_id:
3233                     return True
3234
3235             return False
3236
3237         def get_scale_group_instance_delta(nsr_id, group_name, xact):
3238             delta = {"added": [], "deleted": []}
3239             for instance_cfg, keyspec in self._scale_regh.get_xact_elements(xact, include_keyspec=True):
3240                 elem_nsr_id = nsr_id_from_keyspec(keyspec)
3241                 if elem_nsr_id != nsr_id:
3242                     continue
3243
3244                 elem_group_name = group_name_from_keyspec(keyspec)
3245                 if elem_group_name != group_name:
3246                     continue
3247
3248                 delta["added"].append(instance_cfg.id)
3249
3250             for instance_cfg, keyspec in self._scale_regh.get_xact_elements(include_keyspec=True):
3251                 elem_nsr_id = nsr_id_from_keyspec(keyspec)
3252                 if elem_nsr_id != nsr_id:
3253                     continue
3254
3255                 elem_group_name = group_name_from_keyspec(keyspec)
3256                 if elem_group_name != group_name:
3257                     continue
3258
3259                 if instance_cfg.id in delta["added"]:
3260                     delta["added"].remove(instance_cfg.id)
3261                 else:
3262                     delta["deleted"].append(instance_cfg.id)
3263
3264             return delta
3265
3266         @asyncio.coroutine
3267         def update_nsr_nsd(nsr_id, xact, scratch):
3268
3269             @asyncio.coroutine
3270             def get_nsr_vl_delta(nsr_id, xact, scratch):
3271                 delta = {"added": [], "deleted": []}
3272                 for instance_cfg, keyspec in self._nsr_regh.get_xact_elements(xact, include_keyspec=True):
3273                     elem_nsr_id = nsr_id_from_keyspec(keyspec)
3274                     if elem_nsr_id != nsr_id:
3275                         continue
3276
3277                     if 'vld' in instance_cfg.nsd:
3278                         for vld in instance_cfg.nsd.vld:
3279                             delta["added"].append(vld)
3280
3281                 for instance_cfg, keyspec in self._nsr_regh.get_xact_elements(include_keyspec=True):
3282                     self._log.debug("NSR update: %s", instance_cfg)
3283                     elem_nsr_id = nsr_id_from_keyspec(keyspec)
3284                     if elem_nsr_id != nsr_id:
3285                         continue
3286
3287                     if 'vld' in instance_cfg.nsd:
3288                         for vld in instance_cfg.nsd.vld:
3289                             if vld in delta["added"]:
3290                                 delta["added"].remove(vld)
3291                             else:
3292                                 delta["deleted"].append(vld)
3293
3294                 return delta
3295
3296             vl_delta = yield from get_nsr_vl_delta(nsr_id, xact, scratch)
3297             self._log.debug("Got NSR:%s VL instance delta: %s", nsr_id, vl_delta)
3298
3299             for vld in vl_delta["added"]:
3300                 yield from self._nsm.nsr_instantiate_vl(nsr_id, vld)
3301
3302             for vld in vl_delta["deleted"]:
3303                 yield from self._nsm.nsr_terminate_vl(nsr_id, vld)
3304
3305         def get_nsr_key_pairs(dts_member_reg, xact):
3306             key_pairs = {}
3307             for instance_cfg, keyspec in dts_member_reg.get_xact_elements(xact, include_keyspec=True):
3308                 self._log.debug("Key pair received is {} KS: {}".format(instance_cfg, keyspec))
3309                 xpath = keyspec.to_xpath(RwNsrYang.get_schema())
3310                 key_pairs[instance_cfg.name] = instance_cfg
3311             return key_pairs
3312
3313         def on_apply(dts, acg, xact, action, scratch):
3314             """Apply the  configuration"""
3315             self._log.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3316                             xact, action, scratch)
3317
3318             def handle_create_nsr(msg, key_pairs=None, restart_mode=False):
3319                 # Handle create nsr requests """
3320                 # Do some validations
3321                 if not msg.has_field("nsd"):
3322                     err = "NSD not provided"
3323                     self._log.error(err)
3324                     raise NetworkServiceRecordError(err)
3325
3326                 self._log.debug("Creating NetworkServiceRecord %s  from nsr config  %s",
3327                                msg.id, msg.as_dict())
3328                 nsr = self.nsm.create_nsr(msg, key_pairs=key_pairs, restart_mode=restart_mode)
3329                 return nsr
3330
3331             def handle_delete_nsr(msg):
3332                 @asyncio.coroutine
3333                 def delete_instantiation(ns_id):
3334                     """ Delete instantiation """
3335                     with self._dts.transaction() as xact:
3336                         yield from self._nsm.terminate_ns(ns_id, xact)
3337
3338                 # Handle delete NSR requests
3339                 self._log.info("Delete req for  NSR Id: %s received", msg.id)
3340                 # Terminate the NSR instance
3341                 nsr = self._nsm.get_ns_by_nsr_id(msg.id)
3342
3343                 nsr.set_state(NetworkServiceRecordState.TERMINATE_RCVD)
3344                 event_descr = "Terminate rcvd for NS Id:%s" % msg.id
3345                 nsr.record_event("terminate-rcvd", event_descr)
3346
3347                 self._loop.create_task(delete_instantiation(msg.id))
3348
3349             @asyncio.coroutine
3350             def begin_instantiation(nsr):
3351                 # Begin instantiation
3352                 self._log.info("Beginning NS instantiation: %s", nsr.id)
3353                 try:
3354                     yield from self._nsm.instantiate_ns(nsr.id, xact)
3355                 except Exception as e:
3356                     self._log.exception("NS instantiation: {}".format(e))
3357                     raise e
3358
3359             self._log.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3360                             xact, action, scratch)
3361
3362             if action == rwdts.AppconfAction.INSTALL and xact.id is None:
3363                 key_pairs = []
3364                 if self._key_pair_regh:
3365                     for element in self._key_pair_regh.elements:
3366                         key_pairs.append(element)
3367                 else:
3368                     self._log.error("Reg handle none for key pair in project {}".
3369                                     format(self._project))
3370
3371                 if self._nsr_regh:
3372                     for element in self._nsr_regh.elements:
3373                         nsr = handle_create_nsr(element, key_pairs, restart_mode=True)
3374                         self._loop.create_task(begin_instantiation(nsr))
3375                 else:
3376                     self._log.error("Reg handle none for NSR in project {}".
3377                                     format(self._project))
3378
3379
3380             (added_msgs, deleted_msgs, updated_msgs) = get_add_delete_update_cfgs(self._nsr_regh,
3381                                                                                   xact,
3382                                                                                   "id")
3383             self._log.debug("Added: %s, Deleted: %s, Updated: %s", added_msgs,
3384                             deleted_msgs, updated_msgs)
3385
3386             for msg in added_msgs:
3387                 if msg.id not in self._nsm.nsrs:
3388                     self._log.info("Create NSR received in on_apply to instantiate NS:%s", msg.id)
3389                     key_pairs = get_nsr_key_pairs(self._key_pair_regh, xact)
3390                     nsr = handle_create_nsr(msg,key_pairs)
3391                     self._loop.create_task(begin_instantiation(nsr))
3392
3393             for msg in deleted_msgs:
3394                 self._log.info("Delete NSR received in on_apply to terminate NS:%s", msg.id)
3395                 try:
3396                     handle_delete_nsr(msg)
3397                 except Exception:
3398                     self._log.exception("Failed to terminate NS:%s", msg.id)
3399
3400             for msg in updated_msgs:
3401                 self._log.info("Update NSR received in on_apply: %s", msg)
3402
3403                 self._nsm.nsr_update_cfg(msg.id, msg)
3404
3405                 if 'nsd' in msg:
3406                     self._loop.create_task(update_nsr_nsd(msg.id, xact, scratch))
3407
3408                 for group in msg.scaling_group:
3409                     instance_delta = get_scale_group_instance_delta(msg.id, group.scaling_group_name_ref, xact)
3410                     self._log.debug("Got NSR:%s scale group instance delta: %s", msg.id, instance_delta)
3411
3412                     for instance_id in instance_delta["added"]:
3413                         self._nsm.scale_nsr_out(msg.id, group.scaling_group_name_ref, instance_id, xact)
3414
3415                     for instance_id in instance_delta["deleted"]:
3416                         self._nsm.scale_nsr_in(msg.id, group.scaling_group_name_ref, instance_id)
3417
3418
3419             return RwTypes.RwStatus.SUCCESS
3420
3421         @asyncio.coroutine
3422         def on_prepare(dts, acg, xact, xact_info, ks_path, msg, scratch):
3423             """ Prepare calllback from DTS for NSR """
3424
3425             xpath = ks_path.to_xpath(RwNsrYang.get_schema())
3426             action = xact_info.query_action
3427             self._log.debug(
3428                     "Got Nsr prepare callback (xact: %s) (action: %s) (info: %s), %s:%s)",
3429                     xact, action, xact_info, xpath, msg
3430                     )
3431
3432             @asyncio.coroutine
3433             def delete_instantiation(ns_id):
3434                 """ Delete instantiation """
3435                 yield from self._nsm.terminate_ns(ns_id, None)
3436
3437             def handle_delete_nsr():
3438                 """ Handle delete NSR requests """
3439                 self._log.info("Delete req for  NSR Id: %s received", msg.id)
3440                 # Terminate the NSR instance
3441                 nsr = self._nsm.get_ns_by_nsr_id(msg.id)
3442
3443                 nsr.set_state(NetworkServiceRecordState.TERMINATE_RCVD)
3444                 event_descr = "Terminate rcvd for NS Id:%s" % msg.id
3445                 nsr.record_event("terminate-rcvd", event_descr)
3446
3447                 self._loop.create_task(delete_instantiation(msg.id))
3448
3449             fref = ProtobufC.FieldReference.alloc()
3450             fref.goto_whole_message(msg.to_pbcm())
3451
3452             if action in [rwdts.QueryAction.CREATE, rwdts.QueryAction.UPDATE, rwdts.QueryAction.DELETE]:
3453                 # if this is an NSR create
3454                 if action != rwdts.QueryAction.DELETE and msg.id not in self._nsm.nsrs:
3455                     # Ensure the Cloud account/datacenter has been specified
3456                     if not msg.has_field("cloud_account") and not msg.has_field("om_datacenter"):
3457                         raise NsrInstantiationFailed("Cloud account or datacenter not specified in NSR")
3458
3459                     # Check if nsd is specified
3460                     if not msg.has_field("nsd"):
3461                         raise NsrInstantiationFailed("NSD not specified in NSR")
3462
3463                 else:
3464                     nsr = self._nsm.nsrs[msg.id]
3465
3466                     if msg.has_field("nsd"):
3467                         if nsr.state != NetworkServiceRecordState.RUNNING:
3468                             raise NsrVlUpdateError("Unable to update VL when NSR not in running state")
3469                         if 'vld' not in msg.nsd or len(msg.nsd.vld) == 0:
3470                             raise NsrVlUpdateError("NS config NSD should have atleast 1 VLD defined")
3471
3472                     if msg.has_field("scaling_group"):
3473                         if nsr.state != NetworkServiceRecordState.RUNNING:
3474                             raise ScalingOperationError("Unable to perform scaling action when NS is not in running state")
3475
3476                         if len(msg.scaling_group) > 1:
3477                             raise ScalingOperationError("Only a single scaling group can be configured at a time")
3478
3479                         for group_msg in msg.scaling_group:
3480                             num_new_group_instances = len(group_msg.instance)
3481                             if num_new_group_instances > 1:
3482                                 raise ScalingOperationError("Only a single scaling instance can be modified at a time")
3483
3484                             elif num_new_group_instances == 1:
3485                                 scale_group = nsr.scaling_groups[group_msg.scaling_group_name_ref]
3486                                 if action in [rwdts.QueryAction.CREATE, rwdts.QueryAction.UPDATE]:
3487                                     if len(scale_group.instances) == scale_group.max_instance_count:
3488                                         raise ScalingOperationError("Max instances for %s reached" % scale_group)
3489
3490             acg.handle.prepare_complete_ok(xact_info.handle)
3491
3492
3493         xpath = self._project.add_project(NsrDtsHandler.NSR_XPATH)
3494         self._log.debug("Registering for NSR config using xpath: {}".
3495                         format(xpath))
3496
3497         acg_hdl = rift.tasklets.AppConfGroup.Handler(on_apply=on_apply)
3498         with self._dts.appconf_group_create(handler=acg_hdl) as acg:
3499             self._nsr_regh = acg.register(
3500                 xpath=xpath,
3501                 flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY | rwdts.Flag.CACHE,
3502                 on_prepare=on_prepare
3503             )
3504
3505             self._scale_regh = acg.register(
3506                 xpath=self._project.add_project(NsrDtsHandler.SCALE_INSTANCE_XPATH),
3507                 flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY| rwdts.Flag.CACHE,
3508             )
3509
3510             self._key_pair_regh = acg.register(
3511                 xpath=self._project.add_project(NsrDtsHandler.KEY_PAIR_XPATH),
3512                 flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY | rwdts.Flag.CACHE,
3513             )
3514
3515     def deregister(self):
3516         self._log.debug("De-register NSR config for project {}".
3517                         format(self._project.name))
3518         if self._nsr_regh:
3519             self._nsr_regh.deregister()
3520             self._nsr_regh = None
3521         if self._scale_regh:
3522             self._scale_regh.deregister()
3523             self._scale_regh = None
3524         if self._key_pair_regh:
3525             self._key_pair_regh.deregister()
3526             self._key_pair_regh = None
3527
3528
3529 class NsrOpDataDtsHandler(object):
3530     """ The network service op data DTS handler """
3531     XPATH = "D,/nsr:ns-instance-opdata/nsr:nsr"
3532
3533     def __init__(self, dts, log, loop, nsm):
3534         self._dts = dts
3535         self._log = log
3536         self._loop = loop
3537         self._nsm = nsm
3538
3539         self._project = nsm._project
3540         self._regh = None
3541
3542     @property
3543     def regh(self):
3544         """ Return the registration handle"""
3545         return self._regh
3546
3547     @property
3548     def nsm(self):
3549         """ Return the NS manager instance """
3550         return self._nsm
3551
3552     @asyncio.coroutine
3553     def register(self):
3554         """ Register for Nsr op data publisher registration"""
3555         if self._regh:
3556             self._log.warning("NSR op data handler already registered for project {}".
3557                               format(self._project.name))
3558             return
3559
3560         xpath = self._project.add_project(NsrOpDataDtsHandler.XPATH)
3561         self._log.debug("Registering Nsr op data path {} as publisher".
3562                         format(xpath))
3563
3564         hdl = rift.tasklets.DTS.RegistrationHandler()
3565         handlers = rift.tasklets.Group.Handler()
3566         with self._dts.group_create(handler=handlers) as group:
3567             self._regh = group.register(xpath=xpath,
3568                                         handler=hdl,
3569                                         flags=rwdts.Flag.PUBLISHER | rwdts.Flag.NO_PREP_READ | rwdts.Flag.DATASTORE)
3570
3571     def deregister(self):
3572         self._log.debug("De-register NSR opdata for project {}".
3573                         format(self._project.name))
3574         if self._regh:
3575             self._regh.deregister()
3576             self._regh = None
3577
3578     @asyncio.coroutine
3579     def create(self, xpath, msg):
3580         """
3581         Create an NS record in DTS with the path and message
3582         """
3583         path = self._project.add_project(xpath)
3584         self._log.debug("Creating NSR %s:%s", path, msg)
3585         self.regh.create_element(path, msg)
3586         self._log.debug("Created NSR, %s:%s", path, msg)
3587
3588     @asyncio.coroutine
3589     def update(self, xpath, msg, flags=rwdts.XactFlag.REPLACE):
3590         """
3591         Update an NS record in DTS with the path and message
3592         """
3593         path = self._project.add_project(xpath)
3594         self._log.debug("Updating NSR, %s:%s regh = %s", path, msg, self.regh)
3595         self.regh.update_element(path, msg, flags)
3596         self._log.debug("Updated NSR, %s:%s", path, msg)
3597
3598     @asyncio.coroutine
3599     def delete(self, xpath):
3600         """
3601         Update an NS record in DTS with the path and message
3602         """
3603         path = self._project.add_project(xpath)
3604         self._log.debug("Deleting NSR path:%s", path)
3605         self.regh.delete_element(path)
3606         self._log.debug("Deleted NSR path:%s", path)
3607
3608
3609 class VnfrDtsHandler(object):
3610     """ The virtual network service DTS handler """
3611     XPATH = "D,/vnfr:vnfr-catalog/vnfr:vnfr"
3612
3613     def __init__(self, dts, log, loop, nsm):
3614         self._dts = dts
3615         self._log = log
3616         self._loop = loop
3617         self._nsm = nsm
3618
3619         self._regh = None
3620
3621     @property
3622     def regh(self):
3623         """ Return registration handle """
3624         return self._regh
3625
3626     @property
3627     def nsm(self):
3628         """ Return the NS manager instance """
3629         return self._nsm
3630
3631     @asyncio.coroutine
3632     def register(self):
3633         """ Register for vnfr create/update/delete/ advises from dts """
3634         if self._regh:
3635             self._log.warning("VNFR DTS handler already registered for project {}".
3636                               format(self._project.name))
3637             return
3638
3639
3640         def on_commit(xact_info):
3641             """ The transaction has been committed """
3642             self._log.debug("Got vnfr commit (xact_info: %s)", xact_info)
3643             return rwdts.MemberRspCode.ACTION_OK
3644
3645         @asyncio.coroutine
3646         def on_prepare(xact_info, action, ks_path, msg):
3647             """ prepare callback from dts """
3648             xpath = ks_path.to_xpath(RwNsrYang.get_schema())
3649             self._log.debug(
3650                 "Got vnfr on_prepare cb (xact_info: %s, action: %s): %s:%s",
3651                 xact_info, action, ks_path, msg
3652                 )
3653
3654             schema = VnfrYang.YangData_RwProject_Project_VnfrCatalog_Vnfr.schema()
3655             path_entry = schema.keyspec_to_entry(ks_path)
3656             if path_entry.key00.id not in self._nsm._vnfrs:
3657                 # Check if this is a monitoring param xpath
3658                 if 'vnfr:monitoring-param' not in xpath:
3659                     self._log.error("%s request for non existent record path %s",
3660                                     action, xpath)
3661                 xact_info.respond_xpath(rwdts.XactRspCode.NA, xpath)
3662
3663                 return
3664
3665             if action == rwdts.QueryAction.CREATE or action == rwdts.QueryAction.UPDATE:
3666                 yield from self._nsm.update_vnfr(msg)
3667             elif action == rwdts.QueryAction.DELETE:
3668                 self._log.debug("Deleting VNFR with id %s", path_entry.key00.id)
3669                 self._nsm.delete_vnfr(path_entry.key00.id)
3670
3671             xact_info.respond_xpath(rwdts.XactRspCode.ACK, xpath)
3672
3673         self._log.debug("Registering for VNFR using xpath: %s",
3674                         VnfrDtsHandler.XPATH,)
3675
3676         hdl = rift.tasklets.DTS.RegistrationHandler(on_commit=on_commit,
3677                                                     on_prepare=on_prepare,)
3678         with self._dts.group_create() as group:
3679             self._regh = group.register(xpath=self._nsm._project.add_project(
3680                 VnfrDtsHandler.XPATH),
3681                                         handler=hdl,
3682                                         flags=(rwdts.Flag.SUBSCRIBER),)
3683
3684     def deregister(self):
3685         self._log.debug("De-register VNFR for project {}".
3686                         format(self._nsm._project.name))
3687         if self._regh:
3688             self._regh.deregister()
3689             self._regh = None
3690
3691 class NsdRefCountDtsHandler(object):
3692     """ The NSD Ref Count DTS handler """
3693     XPATH = "D,/nsr:ns-instance-opdata/rw-nsr:nsd-ref-count"
3694
3695     def __init__(self, dts, log, loop, nsm):
3696         self._dts = dts
3697         self._log = log
3698         self._loop = loop
3699         self._nsm = nsm
3700
3701         self._regh = None
3702
3703     @property
3704     def regh(self):
3705         """ Return registration handle """
3706         return self._regh
3707
3708     @property
3709     def nsm(self):
3710         """ Return the NS manager instance """
3711         return self._nsm
3712
3713     @asyncio.coroutine
3714     def register(self):
3715         """ Register for NSD ref count read from dts """
3716         if self._regh:
3717             self._log.warning("NSD ref DTS handler already registered for project {}".
3718                               format(self._project.name))
3719             return
3720
3721
3722         @asyncio.coroutine
3723         def on_prepare(xact_info, action, ks_path, msg):
3724             """ prepare callback from dts """
3725             xpath = ks_path.to_xpath(RwNsrYang.get_schema())
3726
3727             if action == rwdts.QueryAction.READ:
3728                 schema = RwNsrYang.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount.schema()
3729                 path_entry = schema.keyspec_to_entry(ks_path)
3730                 nsd_list = yield from self._nsm.get_nsd_refcount(path_entry.key00.nsd_id_ref)
3731                 for xpath, msg in nsd_list:
3732                     xact_info.respond_xpath(rsp_code=rwdts.XactRspCode.MORE,
3733                                             xpath=xpath,
3734                                             msg=msg)
3735                 xact_info.respond_xpath(rwdts.XactRspCode.ACK)
3736             else:
3737                 raise NetworkServiceRecordError("Not supported operation %s" % action)
3738
3739         hdl = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_prepare,)
3740         with self._dts.group_create() as group:
3741             self._regh = group.register(xpath=self._nsm._project.add_project(
3742                 NsdRefCountDtsHandler.XPATH),
3743                                         handler=hdl,
3744                                         flags=rwdts.Flag.PUBLISHER,)
3745
3746     def deregister(self):
3747         self._log.debug("De-register NSD Ref count for project {}".
3748                         format(self._nsm._project.name))
3749         if self._regh:
3750             self._regh.deregister()
3751             self._regh = None
3752
3753
3754 class NsManager(object):
3755     """ The Network Service Manager class"""
3756     def __init__(self, dts, log, loop, project,
3757                  nsr_handler, vnfr_handler, vlr_handler, ro_plugin_selector,
3758                  vnffgmgr, vnfd_pub_handler, cloud_account_handler):
3759         self._dts = dts
3760         self._log = log
3761         self._loop = loop
3762         self._project = project
3763         self._nsr_handler = nsr_handler
3764         self._vnfr_pub_handler = vnfr_handler
3765         self._vlr_pub_handler = vlr_handler
3766         self._vnffgmgr = vnffgmgr
3767         self._vnfd_pub_handler = vnfd_pub_handler
3768         self._cloud_account_handler = cloud_account_handler
3769
3770         self._ro_plugin_selector = ro_plugin_selector
3771         self._ncclient = rift.mano.ncclient.NcClient(
3772               host="127.0.0.1",
3773               port=2022,
3774               username="admin",
3775               password="admin",
3776               loop=self._loop)
3777
3778         self._nsrs = {}
3779         self._nsds = {}
3780         self._vnfds = {}
3781         self._vnfrs = {}
3782
3783         self.cfgmgr_obj = conman.ROConfigManager(log, loop, dts, self)
3784
3785         # TODO: All these handlers should move to tasklet level.
3786         # Passing self is often an indication of bad design
3787         self._nsd_dts_handler = NsdDtsHandler(dts, log, loop, self)
3788         self._vnfd_dts_handler = VnfdDtsHandler(dts, log, loop, self)
3789         self._dts_handlers = [self._nsd_dts_handler,
3790                               VnfrDtsHandler(dts, log, loop, self),
3791                               NsdRefCountDtsHandler(dts, log, loop, self),
3792                               NsrDtsHandler(dts, log, loop, self),
3793                               ScalingRpcHandler(log, dts, loop, self._project,
3794                                                 self.scale_rpc_callback),
3795                               NsrRpcDtsHandler(dts, log, loop, self),
3796                               self._vnfd_dts_handler,
3797                               self.cfgmgr_obj,
3798                               ]
3799
3800
3801     @property
3802     def log(self):
3803         """ Log handle """
3804         return self._log
3805
3806     @property
3807     def loop(self):
3808         """ Loop """
3809         return self._loop
3810
3811     @property
3812     def dts(self):
3813         """ DTS handle """
3814         return self._dts
3815
3816     @property
3817     def nsr_handler(self):
3818         """" NSR handler """
3819         return self._nsr_handler
3820
3821     @property
3822     def so_obj(self):
3823         """" So Obj handler """
3824         return self._so_obj
3825
3826     @property
3827     def nsrs(self):
3828         """ NSRs in this NSM"""
3829         return self._nsrs
3830
3831     @property
3832     def nsds(self):
3833         """ NSDs in this NSM"""
3834         return self._nsds
3835
3836     @property
3837     def vnfds(self):
3838         """ VNFDs in this NSM"""
3839         return self._vnfds
3840
3841     @property
3842     def vnfrs(self):
3843         """ VNFRs in this NSM"""
3844         return self._vnfrs
3845
3846     @property
3847     def nsr_pub_handler(self):
3848         """ NSR publication handler """
3849         return self._nsr_handler
3850
3851     @property
3852     def vnfr_pub_handler(self):
3853         """ VNFR publication handler """
3854         return self._vnfr_pub_handler
3855
3856     @property
3857     def vlr_pub_handler(self):
3858         """ VLR publication handler """
3859         return self._vlr_pub_handler
3860
3861     @property
3862     def vnfd_pub_handler(self):
3863         return self._vnfd_pub_handler
3864
3865     @asyncio.coroutine
3866     def register(self):
3867         """ Register all static DTS handlers """
3868         for dts_handle in self._dts_handlers:
3869             yield from dts_handle.register()
3870
3871     def deregister(self):
3872         """ Register all static DTS handlers """
3873         for dts_handle in self._dts_handlers:
3874             dts_handle.deregister()
3875
3876
3877     def get_ns_by_nsr_id(self, nsr_id):
3878         """ get NSR by nsr id """
3879         if nsr_id not in self._nsrs:
3880             raise NetworkServiceRecordError("NSR id %s not found" % nsr_id)
3881
3882         return self._nsrs[nsr_id]
3883
3884     def scale_nsr_out(self, nsr_id, scale_group_name, instance_id, config_xact):
3885         self.log.debug("Scale out NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3886                        nsr_id,
3887                        scale_group_name,
3888                        instance_id
3889                        )
3890         nsr = self._nsrs[nsr_id]
3891         if nsr.state != NetworkServiceRecordState.RUNNING:
3892             raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3893
3894         self._loop.create_task(nsr.create_scale_group_instance(scale_group_name, instance_id, config_xact))
3895
3896     def scale_nsr_in(self, nsr_id, scale_group_name, instance_id):
3897         self.log.debug("Scale in NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3898                        nsr_id,
3899                        scale_group_name,
3900                        instance_id,
3901                        )
3902         nsr = self._nsrs[nsr_id]
3903         if nsr.state != NetworkServiceRecordState.RUNNING:
3904             raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3905
3906         self._loop.create_task(nsr.delete_scale_group_instance(scale_group_name, instance_id))
3907
3908     def scale_rpc_callback(self, xact, msg, action):
3909         """Callback handler for RPC calls
3910         Args:
3911             xact : Transaction Handler
3912             msg : RPC input
3913             action : Scaling Action
3914         """
3915         ScalingGroupInstance = NsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup_Instance
3916         ScalingGroup = NsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup
3917
3918         xpath = self._project.add_project(
3919             ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').
3920             format(msg.nsr_id_ref))
3921
3922         instance = ScalingGroupInstance.from_dict({
3923             "id": msg.instance_id,
3924             "project_name": self._project.name,})
3925
3926         @asyncio.coroutine
3927         def get_nsr_scaling_group():
3928             results = yield from self._dts.query_read(xpath, rwdts.XactFlag.MERGE)
3929
3930             for result in results:
3931                 res = yield from result
3932                 nsr_config = res.result
3933
3934             for scaling_group in nsr_config.scaling_group:
3935                 if scaling_group.scaling_group_name_ref == msg.scaling_group_name_ref:
3936                     break
3937             else:
3938                 scaling_group = nsr_config.scaling_group.add()
3939                 scaling_group.scaling_group_name_ref = msg.scaling_group_name_ref
3940
3941             return (nsr_config, scaling_group)
3942
3943         @asyncio.coroutine
3944         def update_config(nsr_config):
3945             xml = self._ncclient.convert_to_xml(RwNsrYang, nsr_config)
3946             xml = '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml)
3947             yield from self._ncclient.connect()
3948             yield from self._ncclient.manager.edit_config(target="running", config=xml, default_operation="replace")
3949
3950         @asyncio.coroutine
3951         def scale_out():
3952             nsr_config, scaling_group = yield from get_nsr_scaling_group()
3953             scaling_group.instance.append(instance)
3954             yield from update_config(nsr_config)
3955
3956         @asyncio.coroutine
3957         def scale_in():
3958             nsr_config, scaling_group = yield from get_nsr_scaling_group()
3959             scaling_group.instance.remove(instance)
3960             yield from update_config(nsr_config)
3961
3962         if action == ScalingRpcHandler.ACTION.SCALE_OUT:
3963             self._loop.create_task(scale_out())
3964         else:
3965             self._loop.create_task(scale_in())
3966
3967         # Opdata based calls, disabled for now!
3968         # if action == ScalingRpcHandler.ACTION.SCALE_OUT:
3969         #     self.scale_nsr_out(
3970         #           msg.nsr_id_ref,
3971         #           msg.scaling_group_name_ref,
3972         #           msg.instance_id,
3973         #           xact)
3974         # else:
3975         #     self.scale_nsr_in(
3976         #           msg.nsr_id_ref,
3977         #           msg.scaling_group_name_ref,
3978         #           msg.instance_id)
3979
3980     def nsr_update_cfg(self, nsr_id, msg):
3981         nsr = self._nsrs[nsr_id]
3982         nsr.nsr_cfg_msg= msg
3983
3984     def nsr_instantiate_vl(self, nsr_id, vld):
3985         self.log.debug("NSR {} create VL {}".format(nsr_id, vld))
3986         nsr = self._nsrs[nsr_id]
3987         if nsr.state != NetworkServiceRecordState.RUNNING:
3988             raise NsrVlUpdateError("Cannot perform VL instantiate if NSR is not in running state")
3989
3990         # Not calling in a separate task as this is called from a separate task
3991         yield from nsr.create_vl_instance(vld)
3992
3993     def nsr_terminate_vl(self, nsr_id, vld):
3994         self.log.debug("NSR {} delete VL {}".format(nsr_id, vld.id))
3995         nsr = self._nsrs[nsr_id]
3996         if nsr.state != NetworkServiceRecordState.RUNNING:
3997             raise NsrVlUpdateError("Cannot perform VL terminate if NSR is not in running state")
3998
3999         # Not calling in a separate task as this is called from a separate task
4000         yield from nsr.delete_vl_instance(vld)
4001
4002     def create_nsr(self, nsr_msg, key_pairs=None,restart_mode=False):
4003         """ Create an NSR instance """
4004         if nsr_msg.id in self._nsrs:
4005             msg = "NSR id %s already exists" % nsr_msg.id
4006             self._log.error(msg)
4007             raise NetworkServiceRecordError(msg)
4008
4009         self._log.info("Create NetworkServiceRecord nsr id %s from nsd_id %s, restart mode %s",
4010                        nsr_msg.id,
4011                        nsr_msg.nsd.id,
4012                        restart_mode)
4013
4014         nsm_plugin = self._ro_plugin_selector.ro_plugin
4015         sdn_account_name = self._cloud_account_handler.get_cloud_account_sdn_name(nsr_msg.cloud_account)
4016
4017         nsr = NetworkServiceRecord(self._dts,
4018                                    self._log,
4019                                    self._loop,
4020                                    self,
4021                                    nsm_plugin,
4022                                    nsr_msg,
4023                                    sdn_account_name,
4024                                    key_pairs,
4025                                    self._project,
4026                                    restart_mode=restart_mode,
4027                                    vlr_handler=self._ro_plugin_selector._records_publisher._vlr_pub_hdlr
4028                                    )
4029         self._nsrs[nsr_msg.id] = nsr
4030         nsm_plugin.create_nsr(nsr_msg, nsr_msg.nsd, key_pairs)
4031
4032         return nsr
4033
4034     def delete_nsr(self, nsr_id):
4035         """
4036         Delete NSR with the passed nsr id
4037         """
4038         del self._nsrs[nsr_id]
4039
4040     @asyncio.coroutine
4041     def instantiate_ns(self, nsr_id, config_xact):
4042         """ Instantiate an NS instance """
4043         self._log.debug("Instantiating Network service id %s", nsr_id)
4044         if nsr_id not in self._nsrs:
4045             err = "NSR id %s not found " % nsr_id
4046             self._log.error(err)
4047             raise NetworkServiceRecordError(err)
4048
4049         nsr = self._nsrs[nsr_id]
4050         yield from nsr.nsm_plugin.instantiate_ns(nsr, config_xact)
4051
4052     @asyncio.coroutine
4053     def update_vnfr(self, vnfr):
4054         """Create/Update an VNFR """
4055
4056         vnfr_state = self._vnfrs[vnfr.id].state
4057         self._log.debug("Updating VNFR with state %s: vnfr %s", vnfr_state, vnfr)
4058
4059         yield from self._vnfrs[vnfr.id].update_state(vnfr)
4060         nsr = self.find_nsr_for_vnfr(vnfr.id)
4061         yield from nsr.update_state()
4062
4063     def find_nsr_for_vnfr(self, vnfr_id):
4064         """ Find the NSR which )has the passed vnfr id"""
4065         for nsr in list(self.nsrs.values()):
4066             for vnfr in list(nsr.vnfrs.values()):
4067                 if vnfr.id == vnfr_id:
4068                     return nsr
4069         return None
4070
4071     def delete_vnfr(self, vnfr_id):
4072         """ Delete VNFR  with the passed id"""
4073         del self._vnfrs[vnfr_id]
4074
4075     def get_nsd_ref(self, nsd_id):
4076         """ Get network service descriptor for the passed nsd_id
4077             with a reference"""
4078         nsd = self.get_nsd(nsd_id)
4079         nsd.ref()
4080         return nsd
4081
4082     @asyncio.coroutine
4083     def get_nsr_config(self, nsd_id):
4084         xpath = self._project.add_project("C,/nsr:ns-instance-config")
4085         results = yield from self._dts.query_read(xpath, rwdts.XactFlag.MERGE)
4086
4087         for result in results:
4088             entry = yield from result
4089             ns_instance_config = entry.result
4090
4091             for nsr in ns_instance_config.nsr:
4092                 if nsr.nsd.id == nsd_id:
4093                     return nsr
4094
4095         return None
4096
4097     @asyncio.coroutine
4098     def nsd_unref_by_nsr_id(self, nsr_id):
4099         """ Unref the network service descriptor based on NSR id """
4100         self._log.debug("NSR Unref called for Nsr Id:%s", nsr_id)
4101         if nsr_id in self._nsrs:
4102             nsr = self._nsrs[nsr_id]
4103
4104             try:
4105                 nsd = self.get_nsd(nsr.nsd_id)
4106                 self._log.debug("Releasing ref on NSD %s held by NSR %s - Curr %d",
4107                                 nsd.id, nsr.id, nsd.ref_count)
4108                 nsd.unref()
4109             except NetworkServiceDescriptorError:
4110                 # We store a copy of NSD in NSR and the NSD in nsd-catalog
4111                 # could be deleted
4112                 pass
4113
4114         else:
4115             self._log.error("Cannot find NSR with id %s", nsr_id)
4116             raise NetworkServiceDescriptorUnrefError("No NSR with id" % nsr_id)
4117
4118     @asyncio.coroutine
4119     def nsd_unref(self, nsd_id):
4120         """ Unref the network service descriptor associated with the id """
4121         nsd = self.get_nsd(nsd_id)
4122         nsd.unref()
4123
4124     def get_nsd(self, nsd_id):
4125         """ Get network service descriptor for the passed nsd_id"""
4126         if nsd_id not in self._nsds:
4127             self._log.error("Cannot find NSD id:%s", nsd_id)
4128             raise NetworkServiceDescriptorError("Cannot find NSD id:%s", nsd_id)
4129
4130         return self._nsds[nsd_id]
4131
4132     def create_nsd(self, nsd_msg):
4133         """ Create a network service descriptor """
4134         self._log.debug("Create network service descriptor - %s", nsd_msg)
4135         if nsd_msg.id in self._nsds:
4136             self._log.error("Cannot create NSD %s -NSD ID already exists", nsd_msg)
4137             raise NetworkServiceDescriptorError("NSD already exists-%s", nsd_msg.id)
4138
4139         nsd = NetworkServiceDescriptor(
4140                 self._dts,
4141                 self._log,
4142                 self._loop,
4143                 nsd_msg,
4144                 self
4145                 )
4146         self._nsds[nsd_msg.id] = nsd
4147
4148         return nsd
4149
4150     def update_nsd(self, nsd):
4151         """ update the Network service descriptor """
4152         self._log.debug("Update network service descriptor - %s", nsd)
4153         if nsd.id not in self._nsds:
4154             self._log.debug("No NSD found - creating NSD id = %s", nsd.id)
4155             self.create_nsd(nsd)
4156         else:
4157             self._log.debug("Updating NSD id = %s, nsd = %s", nsd.id, nsd)
4158             self._nsds[nsd.id].update(nsd)
4159
4160     def delete_nsd(self, nsd_id):
4161         """ Delete the Network service descriptor with the passed id """
4162         self._log.debug("Deleting the network service descriptor - %s", nsd_id)
4163         if nsd_id not in self._nsds:
4164             self._log.debug("Delete NSD failed - cannot find nsd-id %s", nsd_id)
4165             raise NetworkServiceDescriptorNotFound("Cannot find %s", nsd_id)
4166
4167         if nsd_id not in self._nsds:
4168             self._log.debug("Cannot delete NSD id %s reference exists %s",
4169                             nsd_id,
4170                             self._nsds[nsd_id].ref_count)
4171             raise NetworkServiceDescriptorRefCountExists(
4172                 "Cannot delete :%s, ref_count:%s",
4173                 nsd_id,
4174                 self._nsds[nsd_id].ref_count)
4175
4176         del self._nsds[nsd_id]
4177
4178     def get_vnfd_config(self, xact):
4179         vnfd_dts_reg = self._vnfd_dts_handler.regh
4180         for cfg in vnfd_dts_reg.get_xact_elements(xact):
4181             if cfg.id not in self._vnfds:
4182                 self.create_vnfd(cfg)
4183
4184     def get_vnfd(self, vnfd_id, xact):
4185         """ Get virtual network function descriptor for the passed vnfd_id"""
4186         if vnfd_id not in self._vnfds:
4187             self._log.error("Cannot find VNFD id:%s", vnfd_id)
4188             self.get_vnfd_config(xact)
4189
4190             if vnfd_id not in self._vnfds:
4191                 self._log.error("Cannot find VNFD id:%s", vnfd_id)
4192                 raise VnfDescriptorError("Cannot find VNFD id:%s", vnfd_id)
4193
4194         return self._vnfds[vnfd_id]
4195
4196     def create_vnfd(self, vnfd):
4197         """ Create a virtual network function descriptor """
4198         self._log.debug("Create virtual network function descriptor - %s", vnfd)
4199         if vnfd.id in self._vnfds:
4200             self._log.error("Cannot create VNFD %s -VNFD ID already exists", vnfd)
4201             raise VnfDescriptorError("VNFD already exists-%s", vnfd.id)
4202
4203         self._vnfds[vnfd.id] = vnfd
4204         return self._vnfds[vnfd.id]
4205
4206     def update_vnfd(self, vnfd):
4207         """ Update the virtual network function descriptor """
4208         self._log.debug("Update virtual network function descriptor- %s", vnfd)
4209
4210
4211         if vnfd.id not in self._vnfds:
4212             self._log.debug("No VNFD found - creating VNFD id = %s", vnfd.id)
4213             self.create_vnfd(vnfd)
4214         else:
4215             self._log.debug("Updating VNFD id = %s, vnfd = %s", vnfd.id, vnfd)
4216             self._vnfds[vnfd.id] = vnfd
4217
4218     @asyncio.coroutine
4219     def delete_vnfd(self, vnfd_id):
4220         """ Delete the virtual network function descriptor with the passed id """
4221         self._log.debug("Deleting the virtual network function descriptor - %s", vnfd_id)
4222         if vnfd_id not in self._vnfds:
4223             self._log.debug("Delete VNFD failed - cannot find vnfd-id %s", vnfd_id)
4224             raise VnfDescriptorError("Cannot find %s", vnfd_id)
4225
4226         del self._vnfds[vnfd_id]
4227
4228     def nsd_in_use(self, nsd_id):
4229         """ Is the NSD with the passed id in use """
4230         self._log.debug("Is this NSD in use - msg:%s", nsd_id)
4231         if nsd_id in self._nsds:
4232             return self._nsds[nsd_id].in_use()
4233         return False
4234
4235     @asyncio.coroutine
4236     def publish_nsr(self, xact, path, msg):
4237         """ Publish a NSR """
4238         self._log.debug("Publish NSR with path %s, msg %s",
4239                         path, msg)
4240         yield from self.nsr_handler.update(xact, path, msg)
4241
4242     @asyncio.coroutine
4243     def unpublish_nsr(self, xact, path):
4244         """ Un Publish an NSR """
4245         self._log.debug("Publishing delete NSR with path %s", path)
4246         yield from self.nsr_handler.delete(path, xact)
4247
4248     def vnfr_is_ready(self, vnfr_id):
4249         """ VNFR with the id is ready """
4250         self._log.debug("VNFR id %s ready", vnfr_id)
4251         if vnfr_id not in self._vnfds:
4252             err = "Did not find VNFR ID with id %s" % vnfr_id
4253             self._log.critical("err")
4254             raise VirtualNetworkFunctionRecordError(err)
4255         self._vnfrs[vnfr_id].is_ready()
4256
4257     @asyncio.coroutine
4258     def get_nsd_refcount(self, nsd_id):
4259         """ Get the nsd_list from this NSM"""
4260
4261         def nsd_refcount_xpath(nsd_id):
4262             """ xpath for ref count entry """
4263             return (self._project.add_project(NsdRefCountDtsHandler.XPATH) +
4264                     "[rw-nsr:nsd-id-ref = '{}']").format(nsd_id)
4265
4266         nsd_list = []
4267         if nsd_id is None or nsd_id == "":
4268             for nsd in self._nsds.values():
4269                 nsd_msg = RwNsrYang.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount()
4270                 nsd_msg.nsd_id_ref = nsd.id
4271                 nsd_msg.instance_ref_count = nsd.ref_count
4272                 nsd_list.append((nsd_refcount_xpath(nsd.id), nsd_msg))
4273         elif nsd_id in self._nsds:
4274             nsd_msg = RwNsrYang.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount()
4275             nsd_msg.nsd_id_ref = self._nsds[nsd_id].id
4276             nsd_msg.instance_ref_count = self._nsds[nsd_id].ref_count
4277             nsd_list.append((nsd_refcount_xpath(nsd_id), nsd_msg))
4278
4279         return nsd_list
4280
4281     @asyncio.coroutine
4282     def terminate_ns(self, nsr_id, xact):
4283         """
4284         Terminate network service for the given NSR Id
4285         """
4286
4287         # Terminate the instances/networks assocaited with this nw service
4288         self._log.debug("Terminating the network service %s", nsr_id)
4289         try :
4290             yield from self._nsrs[nsr_id].terminate()
4291         except Exception as e:
4292             self.log.exception("Failed to terminate NSR[id=%s]", nsr_id)
4293
4294         # Unref the NSD
4295         yield from self.nsd_unref_by_nsr_id(nsr_id)
4296
4297         # Unpublish the NSR record
4298         self._log.debug("Unpublishing the network service %s", nsr_id)
4299         yield from self._nsrs[nsr_id].unpublish(xact)
4300
4301         # Finaly delete the NS instance from this NS Manager
4302         self._log.debug("Deletng the network service %s", nsr_id)
4303         self.delete_nsr(nsr_id)
4304
4305
4306 class NsmRecordsPublisherProxy(object):
4307     """ This class provides a publisher interface that allows plugin objects
4308         to publish NSR/VNFR/VLR"""
4309
4310     def __init__(self, dts, log, loop, project, nsr_pub_hdlr,
4311                  vnfr_pub_hdlr, vlr_pub_hdlr,):
4312         self._dts = dts
4313         self._log = log
4314         self._loop = loop
4315         self._project = project
4316         self._nsr_pub_hdlr = nsr_pub_hdlr
4317         self._vlr_pub_hdlr = vlr_pub_hdlr
4318         self._vnfr_pub_hdlr = vnfr_pub_hdlr
4319
4320     @asyncio.coroutine
4321     def publish_nsr(self, xact, nsr):
4322         """ Publish an NSR """
4323         path = NetworkServiceRecord.xpath_from_nsr(nsr)
4324         return (yield from self._nsr_pub_hdlr.update(xact, path, nsr))
4325
4326     @asyncio.coroutine
4327     def unpublish_nsr(self, xact, nsr):
4328         """ Unpublish an NSR """
4329         path = NetworkServiceRecord.xpath_from_nsr(nsr)
4330         return (yield from self._nsr_pub_hdlr.delete(xact, path))
4331
4332     @asyncio.coroutine
4333     def publish_vnfr(self, xact, vnfr):
4334         """ Publish an VNFR """
4335         path = VirtualNetworkFunctionRecord.vnfr_xpath(vnfr)
4336         return (yield from self._vnfr_pub_hdlr.update(xact, path, vnfr))
4337
4338     @asyncio.coroutine
4339     def unpublish_vnfr(self, xact, vnfr):
4340         """ Unpublish a VNFR """
4341         path = VirtualNetworkFunctionRecord.vnfr_xpath(vnfr)
4342         return (yield from self._vnfr_pub_hdlr.delete(xact, path))
4343
4344     @asyncio.coroutine
4345     def publish_vlr(self, xact, vlr):
4346         """ Publish a VLR """
4347         path = self._project.add_project(VirtualLinkRecord.vlr_xpath(vlr))
4348         return (yield from self._vlr_pub_hdlr.update(xact, path, vlr))
4349
4350     @asyncio.coroutine
4351     def unpublish_vlr(self, xact, vlr):
4352         """ Unpublish a VLR """
4353         path = self._project.add_project(VirtualLinkRecord.vlr_xpath(vlr))
4354         return (yield from self._vlr_pub_hdlr.delete(xact, path))
4355
4356
4357 class ScalingRpcHandler(mano_dts.DtsHandler):
4358     """ The Network service Monitor DTS handler """
4359     SCALE_IN_INPUT_XPATH = "I,/nsr:exec-scale-in"
4360     SCALE_IN_OUTPUT_XPATH = "O,/nsr:exec-scale-in"
4361
4362     SCALE_OUT_INPUT_XPATH = "I,/nsr:exec-scale-out"
4363     SCALE_OUT_OUTPUT_XPATH = "O,/nsr:exec-scale-out"
4364
4365     ACTION = Enum('ACTION', 'SCALE_IN SCALE_OUT')
4366
4367     def __init__(self, log, dts, loop, project, callback=None):
4368         super().__init__(log, dts, loop, project)
4369         self.callback = callback
4370         self.last_instance_id = defaultdict(int)
4371         self._regh_in = None
4372         self._regh_out = None
4373
4374     @asyncio.coroutine
4375     def register(self):
4376
4377         if self._regh_in:
4378             self._log.warning("RPC already registered for project {}".
4379                               format(self._project.name))
4380             return
4381
4382         @asyncio.coroutine
4383         def on_scale_in_prepare(xact_info, action, ks_path, msg):
4384             assert action == rwdts.QueryAction.RPC
4385
4386             try:
4387                 if not self._project.rpc_check(msg, xact_info=xact_info):
4388                     return
4389
4390                 if self.callback:
4391                     self.callback(xact_info.xact, msg, self.ACTION.SCALE_IN)
4392
4393                 rpc_op = NsrYang.YangOutput_Nsr_ExecScaleIn.from_dict({
4394                     "instance_id": msg.instance_id,
4395                     "project_name": self._project.name,})
4396
4397                 xact_info.respond_xpath(
4398                     rwdts.XactRspCode.ACK,
4399                     self.__class__.SCALE_IN_OUTPUT_XPATH,
4400                     rpc_op)
4401
4402             except Exception as e:
4403                 self.log.exception(e)
4404                 xact_info.respond_xpath(
4405                     rwdts.XactRspCode.NACK,
4406                     self.__class__.SCALE_IN_OUTPUT_XPATH)
4407
4408         @asyncio.coroutine
4409         def on_scale_out_prepare(xact_info, action, ks_path, msg):
4410             assert action == rwdts.QueryAction.RPC
4411
4412             try:
4413                 if not self._project.rpc_check(msg, xact_info=xact_info):
4414                     return
4415
4416                 scaling_group = msg.scaling_group_name_ref
4417                 if not msg.instance_id:
4418                     last_instance_id = self.last_instance_id[scale_group]
4419                     msg.instance_id  = last_instance_id + 1
4420                     self.last_instance_id[scale_group] += 1
4421
4422                 if self.callback:
4423                     self.callback(xact_info.xact, msg, self.ACTION.SCALE_OUT)
4424
4425                 rpc_op = NsrYang.YangOutput_Nsr_ExecScaleOut.from_dict({
4426                     "instance_id": msg.instance_id,
4427                     "project_name": self._project.name,})
4428
4429                 xact_info.respond_xpath(
4430                     rwdts.XactRspCode.ACK,
4431                     self.__class__.SCALE_OUT_OUTPUT_XPATH,
4432                     rpc_op)
4433
4434             except Exception as e:
4435                 self.log.exception(e)
4436                 xact_info.respond_xpath(
4437                       rwdts.XactRspCode.NACK,
4438                       self.__class__.SCALE_OUT_OUTPUT_XPATH)
4439
4440         scale_in_hdl = rift.tasklets.DTS.RegistrationHandler(
4441               on_prepare=on_scale_in_prepare)
4442         scale_out_hdl = rift.tasklets.DTS.RegistrationHandler(
4443               on_prepare=on_scale_out_prepare)
4444
4445         with self.dts.group_create() as group:
4446             self._regh_in = group.register(
4447                 xpath=self.__class__.SCALE_IN_INPUT_XPATH,
4448                 handler=scale_in_hdl,
4449                 flags=rwdts.Flag.PUBLISHER)
4450             self._regh_out = group.register(
4451                 xpath=self.__class__.SCALE_OUT_INPUT_XPATH,
4452                 handler=scale_out_hdl,
4453                 flags=rwdts.Flag.PUBLISHER)
4454
4455     def deregister(self):
4456         self._log.debug("De-register scale RPCs for project {}".
4457                         format(self._project.name))
4458         if self._regh_in:
4459             self._regh_in.deregister()
4460             self._regh_in = None
4461         if self._regh_out:
4462             self._regh_out.deregister()
4463             self._regh_out = None
4464
4465
4466 class NsmProject(ManoProject):
4467
4468     def __init__(self, name, tasklet, **kw):
4469         super(NsmProject, self).__init__(tasklet.log, name)
4470         self.update(tasklet)
4471
4472         self._nsm = None
4473
4474         self._ro_plugin_selector = None
4475         self._vnffgmgr = None
4476
4477         self._nsr_pub_handler = None
4478         self._vnfr_pub_handler = None
4479         self._vlr_pub_handler = None
4480         self._vnfd_pub_handler = None
4481         self._scale_cfg_handler = None
4482
4483         self._records_publisher_proxy = None
4484
4485     @asyncio.coroutine
4486     def register(self):
4487         self._nsr_pub_handler = publisher.NsrOpDataDtsHandler(
4488             self._dts, self.log, self.loop, self)
4489         yield from self._nsr_pub_handler.register()
4490
4491         self._vnfr_pub_handler = publisher.VnfrPublisherDtsHandler(
4492             self._dts, self.log, self.loop, self)
4493         yield from self._vnfr_pub_handler.register()
4494
4495         self._vlr_pub_handler = publisher.VlrPublisherDtsHandler(
4496             self._dts, self.log, self.loop, self)
4497         yield from self._vlr_pub_handler.register()
4498
4499         manifest = self._tasklet.tasklet_info.get_pb_manifest()
4500         use_ssl = manifest.bootstrap_phase.rwsecurity.use_ssl
4501         ssl_cert = manifest.bootstrap_phase.rwsecurity.cert
4502         ssl_key = manifest.bootstrap_phase.rwsecurity.key
4503
4504         self._vnfd_pub_handler = publisher.VnfdPublisher(
4505             use_ssl, ssl_cert, ssl_key, self.loop, self)
4506
4507         self._records_publisher_proxy = NsmRecordsPublisherProxy(
4508                 self._dts,
4509                 self.log,
4510                 self.loop,
4511                 self,
4512                 self._nsr_pub_handler,
4513                 self._vnfr_pub_handler,
4514                 self._vlr_pub_handler,
4515                 )
4516
4517         # Register the NSM to receive the nsm plugin
4518         # when cloud account is configured
4519         self._ro_plugin_selector = cloud.ROAccountPluginSelector(
4520                 self._dts,
4521                 self.log,
4522                 self.loop,
4523                 self,
4524                 self._records_publisher_proxy,
4525                 )
4526         yield from self._ro_plugin_selector.register()
4527
4528         self._cloud_account_handler = cloud.CloudAccountConfigSubscriber(
4529                 self._log,
4530                 self._dts,
4531                 self.log_hdl,
4532                 self,
4533                 )
4534
4535         yield from self._cloud_account_handler.register()
4536
4537         self._vnffgmgr = rwvnffgmgr.VnffgMgr(self._dts, self.log, self.log_hdl, self.loop, self)
4538         yield from self._vnffgmgr.register()
4539
4540         self._nsm = NsManager(
4541                 self._dts,
4542                 self.log,
4543                 self.loop,
4544                 self,
4545                 self._nsr_pub_handler,
4546                 self._vnfr_pub_handler,
4547                 self._vlr_pub_handler,
4548                 self._ro_plugin_selector,
4549                 self._vnffgmgr,
4550                 self._vnfd_pub_handler,
4551                 self._cloud_account_handler,
4552                 )
4553
4554         yield from self._nsm.register()
4555
4556     def deregister(self):
4557         self._log.debug("Project {} de-register".format(self.name))
4558         self._nsm.deregister()
4559         self._vnffgmgr.deregister()
4560         self._cloud_account_handler.deregister()
4561         self._ro_plugin_selector.deregister()
4562         self._nsm = None
4563
4564     @asyncio.coroutine
4565     def delete_prepare(self):
4566         # Check if any NS instance is present
4567         if self._nsm and self._nsm._nsrs:
4568             return False
4569         return True
4570
4571
4572 class NsmTasklet(rift.tasklets.Tasklet):
4573     """
4574     The network service manager  tasklet
4575     """
4576     def __init__(self, *args, **kwargs):
4577         super(NsmTasklet, self).__init__(*args, **kwargs)
4578         self.rwlog.set_category("rw-mano-log")
4579         self.rwlog.set_subcategory("nsm")
4580
4581         self._dts = None
4582         self.project_handler = None
4583         self.projects = {}
4584
4585     @property
4586     def dts(self):
4587         return self._dts
4588
4589     def start(self):
4590         """ The task start callback """
4591         super(NsmTasklet, self).start()
4592         self.log.info("Starting NsmTasklet")
4593
4594         self.log.debug("Registering with dts")
4595         self._dts = rift.tasklets.DTS(self.tasklet_info,
4596                                       RwNsmYang.get_schema(),
4597                                       self.loop,
4598                                       self.on_dts_state_change)
4599
4600         self.log.debug("Created DTS Api GI Object: %s", self._dts)
4601
4602     def stop(self):
4603         try:
4604             self._dts.deinit()
4605         except Exception:
4606             print("Caught Exception in NSM stop:", sys.exc_info()[0])
4607             raise
4608
4609     def on_instance_started(self):
4610         """ Task instance started callback """
4611         self.log.debug("Got instance started callback")
4612
4613     @asyncio.coroutine
4614     def init(self):
4615         """ Task init callback """
4616         self.log.debug("Got instance started callback")
4617
4618         self.log.debug("creating project handler")
4619         self.project_handler = ProjectHandler(self, NsmProject)
4620         self.project_handler.register()
4621
4622
4623
4624     @asyncio.coroutine
4625     def run(self):
4626         """ Task run callback """
4627         pass
4628
4629     @asyncio.coroutine
4630     def on_dts_state_change(self, state):
4631         """Take action according to current dts state to transition
4632         application into the corresponding application state
4633
4634         Arguments
4635             state - current dts state
4636         """
4637         switch = {
4638             rwdts.State.INIT: rwdts.State.REGN_COMPLETE,
4639             rwdts.State.CONFIG: rwdts.State.RUN,
4640         }
4641
4642         handlers = {
4643             rwdts.State.INIT: self.init,
4644             rwdts.State.RUN: self.run,
4645         }
4646
4647         # Transition application to next state
4648         handler = handlers.get(state, None)
4649         if handler is not None:
4650             yield from handler()
4651
4652         # Transition dts to next state
4653         next_state = switch.get(state, None)
4654         if next_state is not None:
4655             self.log.debug("Changing state to %s", next_state)
4656             self._dts.handle.set_state(next_state)