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",
2060                        vnfd_msg.name,
2061                        const_vnfd.member_vnf_index,
2062                        [ group.name for group in placement_groups])
2063         vnfr = yield from VirtualNetworkFunctionRecord.create_record(self._dts,
2064                                             self._log,
2065                                             self._loop,
2066                                             self._project,
2067                                             vnfd_msg,
2068                                             const_vnfd,
2069                                             self.nsd_id,
2070                                             self.name,
2071                                             cloud_account_name,
2072                                             om_datacenter_name,
2073                                             self.id,
2074                                             group_name,
2075                                             group_instance_id,
2076                                             placement_groups,
2077                                             restart_mode=self.restart_mode,
2078                                             )
2079         if vnfr.id in self._vnfrs:
2080             err = "VNF with VNFR id %s already in vnf list" % (vnfr.id,)
2081             raise NetworkServiceRecordError(err)
2082
2083         self._vnfrs[vnfr.id] = vnfr
2084         self._nsm.vnfrs[vnfr.id] = vnfr
2085
2086         yield from vnfr.set_config_status(NsrYang.ConfigStates.INIT)
2087
2088         self._log.debug("Added VNFR %s to NSM VNFR list with id %s",
2089                         vnfr.name,
2090                         vnfr.id)
2091
2092         return vnfr
2093
2094     def create_param_pools(self):
2095         for param_pool in self.nsd_msg.parameter_pool:
2096             self._log.debug("Found parameter pool %s in nsr id %s", param_pool, self.id)
2097
2098             start_value = param_pool.range.start_value
2099             end_value = param_pool.range.end_value
2100             if end_value < start_value:
2101                 raise NetworkServiceRecordError(
2102                         "Parameter pool %s has invalid range (start: {}, end: {})".format(
2103                             start_value, end_value
2104                             )
2105                         )
2106
2107             self._param_pools[param_pool.name] = config_value_pool.ParameterValuePool(
2108                     self._log,
2109                     param_pool.name,
2110                     range(start_value, end_value)
2111                     )
2112
2113     @asyncio.coroutine
2114     def fetch_vnfr(self, vnfr_path):
2115         """ Fetch VNFR record """
2116         vnfr = None
2117         self._log.debug("Fetching VNFR with key %s while instantiating %s",
2118                         vnfr_path, self.id)
2119         res_iter = yield from self._dts.query_read(vnfr_path, rwdts.XactFlag.MERGE)
2120
2121         for ent in res_iter:
2122             res = yield from ent
2123             vnfr = res.result
2124
2125         return vnfr
2126
2127     @asyncio.coroutine
2128     def instantiate_vnfs(self, vnfrs):
2129         """
2130         This function instantiates VNFs for every VNF in this Network Service
2131         """
2132         self._log.debug("Instantiating %u VNFs in NS %s", len(vnfrs), self.id)
2133         for vnf in vnfrs:
2134             self._log.debug("Instantiating VNF: %s in NS %s", vnf, self.id)
2135             yield from self.nsm_plugin.instantiate_vnf(self, vnf)
2136
2137     @asyncio.coroutine
2138     def instantiate_vnffgs(self):
2139         """
2140         This function instantiates VNFFGs for every VNFFG in this Network Service
2141         """
2142         self._log.debug("Instantiating %u VNFFGs in NS %s",
2143                         len(self.nsd_msg.vnffgd), self.id)
2144         for _, vnfr in self.vnfrs.items():
2145             while vnfr.state in [VnfRecordState.INSTANTIATION_PENDING, VnfRecordState.INIT]:
2146                 self._log.debug("Received vnfr state for vnfr %s is %s; retrying",vnfr.name,vnfr.state)
2147                 yield from asyncio.sleep(2, loop=self._loop)
2148             if vnfr.state == VnfRecordState.ACTIVE:
2149                 self._log.debug("Received vnfr state for vnfr %s is %s ",vnfr.name,vnfr.state)
2150                 continue
2151             else:
2152                 self._log.debug("Received vnfr state for vnfr %s is %s; failing vnffg creation",vnfr.name,vnfr.state)
2153                 self._vnffgr_state = VnffgRecordState.FAILED
2154                 return
2155
2156         self._log.info("Waiting for 90 seconds for VMs to come up")
2157         yield from asyncio.sleep(90, loop=self._loop)
2158         self._log.info("Starting VNFFG orchestration")
2159         for vnffg in self._vnffgrs.values():
2160             self._log.debug("Instantiating VNFFG: %s in NS %s", vnffg, self.id)
2161             yield from vnffg.instantiate()
2162
2163     @asyncio.coroutine
2164     def instantiate_scaling_instances(self, config_xact):
2165         """ Instantiate any default scaling instances in this Network Service """
2166         for group in self._scaling_groups.values():
2167             for i in range(group.min_instance_count):
2168                 self._log.debug("Instantiating %s default scaling instance %s", group, i)
2169                 yield from self.create_scale_group_instance(
2170                         group.name, i, config_xact, is_default=True
2171                         )
2172
2173             for group_msg in self._nsr_cfg_msg.scaling_group:
2174                 if group_msg.scaling_group_name_ref != group.name:
2175                     continue
2176
2177                 for instance in group_msg.instance:
2178                     self._log.debug("Reloading %s scaling instance %s", group_msg, instance.id)
2179                     yield from self.create_scale_group_instance(
2180                             group.name, instance.id, config_xact, is_default=False
2181                             )
2182
2183     def has_scaling_instances(self):
2184         """ Return boolean indicating if the network service has default scaling groups """
2185         for group in self._scaling_groups.values():
2186             if group.min_instance_count > 0:
2187                 return True
2188
2189         for group_msg in self._nsr_cfg_msg.scaling_group:
2190             if len(group_msg.instance) > 0:
2191                 return True
2192
2193         return False
2194
2195     @asyncio.coroutine
2196     def publish(self):
2197         """ This function publishes this NSR """
2198         self._nsr_msg = self.create_msg()
2199
2200         self._log.debug("Publishing the NSR with xpath %s and nsr %s",
2201                         self.nsr_xpath,
2202                         self._nsr_msg)
2203
2204         if self._debug_running:
2205             self._log.debug("Publishing NSR in RUNNING state!")
2206             #raise()
2207
2208         with self._dts.transaction() as xact:
2209             yield from self._nsm.nsr_handler.update(xact, self.nsr_xpath, self._nsr_msg)
2210             if self._op_status.state == NetworkServiceRecordState.RUNNING:
2211                 self._debug_running = True
2212
2213     @asyncio.coroutine
2214     def unpublish(self, xact):
2215         """ Unpublish this NSR object """
2216         self._log.debug("Unpublishing Network service id %s", self.id)
2217         yield from self._nsm.nsr_handler.delete(xact, self.nsr_xpath)
2218
2219     @property
2220     def nsr_xpath(self):
2221         """ Returns the xpath associated with this NSR """
2222         return self._project.add_project((
2223             "D,/nsr:ns-instance-opdata" +
2224             "/nsr:nsr[nsr:ns-instance-config-ref = '{}']"
2225         ).format(self.id))
2226
2227     @staticmethod
2228     def xpath_from_nsr(nsr):
2229         """ Returns the xpath associated with this NSR  op data"""
2230         return self._project.add_project((NetworkServiceRecord.XPATH +
2231                     "[nsr:ns-instance-config-ref = '{}']").format(nsr.id))
2232
2233     @property
2234     def nsd_xpath(self):
2235         """ Return NSD config xpath."""
2236         return self._project.add_project((
2237             "C,/project-nsd:nsd-catalog/project-nsd:nsd[project-nsd:id = '{}']"
2238         ).format(self.nsd_id))
2239
2240     @asyncio.coroutine
2241     def instantiate(self, config_xact):
2242         """"Instantiates a NetworkServiceRecord.
2243
2244         This function instantiates a Network service
2245         which involves the following steps,
2246
2247         * Instantiate every VL in NSD by sending create VLR request to DTS.
2248         * Instantiate every VNF in NSD by sending create VNF reuqest to DTS.
2249         * Publish the NSR details to DTS
2250
2251         Arguments:
2252             nsr:  The NSR configuration request containing nsr-id and nsd
2253             config_xact: The configuration transaction which initiated the instatiation
2254
2255         Raises:
2256             NetworkServiceRecordError if the NSR creation fails
2257
2258         Returns:
2259             No return value
2260         """
2261
2262         self._log.debug("Instantiating NS - %s xact - %s", self, config_xact)
2263
2264         # Move the state to INIITALIZING
2265         self.set_state(NetworkServiceRecordState.INIT)
2266
2267         event_descr = "Instantiation Request Received NSR Id:%s" % self.id
2268         self.record_event("instantiating", event_descr)
2269
2270         # Find the NSD
2271         self._nsd = self._nsr_cfg_msg.nsd
2272
2273         try:
2274             # Update ref count if nsd present in catalog
2275             self._nsm.get_nsd_ref(self.nsd_id)
2276
2277         except NetworkServiceDescriptorError:
2278             # This could be an NSD not in the nsd-catalog
2279             pass
2280
2281         # Merge any config and initial config primitive values
2282         self.config_store.merge_nsd_config(self.nsd_msg)
2283         self._log.debug("Merged NSD: {}".format(self.nsd_msg.as_dict()))
2284
2285         event_descr = "Fetched NSD with descriptor id %s" % self.nsd_id
2286         self.record_event("nsd-fetched", event_descr)
2287
2288         if self._nsd is None:
2289             msg = "Failed to fetch NSD with nsd-id [%s] for nsr-id %s"
2290             self._log.debug(msg, self.nsd_id, self.id)
2291             raise NetworkServiceRecordError(self)
2292
2293         self._log.debug("Got nsd result %s", self._nsd)
2294
2295         # Substitute any input parameters
2296         self.substitute_input_parameters(self._nsd, self._nsr_cfg_msg)
2297
2298         # Create the record
2299         yield from self.create(config_xact)
2300
2301         # Publish the NSR to DTS
2302         yield from self.publish()
2303
2304         @asyncio.coroutine
2305         def do_instantiate():
2306             """
2307                 Instantiate network service
2308             """
2309             self._log.debug("Instantiating VLs nsr id [%s] nsd id [%s]",
2310                             self.id, self.nsd_id)
2311
2312             # instantiate the VLs
2313             event_descr = ("Instantiating %s external VLs for NSR id %s" %
2314                            (len(self.nsd_msg.vld), self.id))
2315             self.record_event("begin-external-vls-instantiation", event_descr)
2316
2317             self.set_state(NetworkServiceRecordState.VL_INIT_PHASE)
2318
2319             yield from self.instantiate_vls()
2320
2321             # Publish the NSR to DTS
2322             yield from self.publish()
2323
2324             event_descr = ("Finished instantiating %s external VLs for NSR id %s" %
2325                            (len(self.nsd_msg.vld), self.id))
2326             self.record_event("end-external-vls-instantiation", event_descr)
2327
2328             self.set_state(NetworkServiceRecordState.VNF_INIT_PHASE)
2329
2330             self._log.debug("Instantiating VNFs  ...... nsr[%s], nsd[%s]",
2331                             self.id, self.nsd_id)
2332
2333             # instantiate the VNFs
2334             event_descr = ("Instantiating %s VNFS for NSR id %s" %
2335                            (len(self.nsd_msg.constituent_vnfd), self.id))
2336
2337             self.record_event("begin-vnf-instantiation", event_descr)
2338
2339             yield from self.instantiate_vnfs(self._vnfrs.values())
2340
2341             self._log.debug(" Finished instantiating %d VNFs for NSR id %s",
2342                             len(self.nsd_msg.constituent_vnfd), self.id)
2343
2344             event_descr = ("Finished instantiating %s VNFs for NSR id %s" %
2345                            (len(self.nsd_msg.constituent_vnfd), self.id))
2346             self.record_event("end-vnf-instantiation", event_descr)
2347
2348             if len(self.vnffgrs) > 0:
2349                 #self.set_state(NetworkServiceRecordState.VNFFG_INIT_PHASE)
2350                 event_descr = ("Instantiating %s VNFFGS for NSR id %s" %
2351                                (len(self.nsd_msg.vnffgd), self.id))
2352
2353                 self.record_event("begin-vnffg-instantiation", event_descr)
2354
2355                 yield from self.instantiate_vnffgs()
2356
2357                 event_descr = ("Finished instantiating %s VNFFGDs for NSR id %s" %
2358                                (len(self.nsd_msg.vnffgd), self.id))
2359                 self.record_event("end-vnffg-instantiation", event_descr)
2360
2361             if self.has_scaling_instances():
2362                 event_descr = ("Instantiating %s Scaling Groups for NSR id %s" %
2363                                (len(self._scaling_groups), self.id))
2364
2365                 self.record_event("begin-scaling-group-instantiation", event_descr)
2366                 yield from self.instantiate_scaling_instances(config_xact)
2367                 self.record_event("end-scaling-group-instantiation", event_descr)
2368
2369             # Give the plugin a chance to deploy the network service now that all
2370             # virtual links and vnfs are instantiated
2371             yield from self.nsm_plugin.deploy(self._nsr_msg)
2372
2373             self._log.debug("Publishing  NSR...... nsr[%s], nsd[%s]",
2374                             self.id, self.nsd_id)
2375
2376             # Publish the NSR to DTS
2377             yield from self.publish()
2378
2379             self._log.debug("Published  NSR...... nsr[%s], nsd[%s]",
2380                             self.id, self.nsd_id)
2381
2382         def on_instantiate_done(fut):
2383             # If the do_instantiate fails, then publish NSR with failed result
2384             if fut.exception() is not None:
2385                 self._log.error("NSR instantiation failed for NSR id %s: %s", self.id, str(fut.exception()))
2386                 self._loop.create_task(self.instantiation_failed(failed_reason=str(fut.exception())))
2387
2388         instantiate_task = self._loop.create_task(do_instantiate())
2389         instantiate_task.add_done_callback(on_instantiate_done)
2390
2391     @asyncio.coroutine
2392     def set_config_status(self, status, status_details=None):
2393         if self.config_status != status:
2394             self._log.debug("Updating NSR {} status for {} to {}".
2395                             format(self.name, self.config_status, status))
2396             self._config_status = status
2397             self._config_status_details = status_details
2398
2399             if self._config_status == NsrYang.ConfigStates.FAILED:
2400                 self.record_event("config-failed", "NS configuration failed",
2401                         evt_details=self._config_status_details)
2402
2403             yield from self.publish()
2404
2405     @asyncio.coroutine
2406     def is_active(self):
2407         """ This NS is active """
2408         self.set_state(NetworkServiceRecordState.RUNNING)
2409         if self._is_active:
2410             return
2411
2412         # Publish the NSR to DTS
2413         self._log.debug("Network service %s is active ", self.id)
2414         self._is_active = True
2415
2416         event_descr = "NSR in running state for NSR id %s" % self.id
2417         self.record_event("ns-running", event_descr)
2418
2419         yield from self.publish()
2420
2421     @asyncio.coroutine
2422     def instantiation_failed(self, failed_reason=None):
2423         """ The NS instantiation failed"""
2424         self._log.error("Network service id:%s, name:%s instantiation failed",
2425                         self.id, self.name)
2426         self.set_state(NetworkServiceRecordState.FAILED)
2427
2428         event_descr = "Instantiation of NS %s failed" % self.id
2429         self.record_event("ns-failed", event_descr, evt_details=failed_reason)
2430
2431         # Publish the NSR to DTS
2432         yield from self.publish()
2433
2434     @asyncio.coroutine
2435     def terminate_vnfrs(self, vnfrs):
2436         """ Terminate VNFRS in this network service """
2437         self._log.debug("Terminating VNFs in network service %s", self.id)
2438         for vnfr in vnfrs:
2439             yield from self.nsm_plugin.terminate_vnf(vnfr)
2440
2441     @asyncio.coroutine
2442     def terminate(self):
2443         """ Terminate a NetworkServiceRecord."""
2444         def terminate_vnffgrs():
2445             """ Terminate VNFFGRS in this network service """
2446             self._log.debug("Terminating VNFFGRs in network service %s", self.id)
2447             for vnffgr in self.vnffgrs.values():
2448                 yield from vnffgr.terminate()
2449
2450         def terminate_vlrs():
2451             """ Terminate VLRs in this netork service """
2452             self._log.debug("Terminating VLs in network service %s", self.id)
2453             for vlr in self.vlrs:
2454                 yield from self.nsm_plugin.terminate_vl(vlr)
2455                 vlr.state = VlRecordState.TERMINATED
2456                 if vlr.id in self.vlr_uptime_tasks:
2457                     self.vlr_uptime_tasks[vlr.id].cancel()
2458
2459         self._log.debug("Terminating network service id %s", self.id)
2460
2461         # Move the state to TERMINATE
2462         self.set_state(NetworkServiceRecordState.TERMINATE)
2463         event_descr = "Terminate being processed for NS Id:%s" % self.id
2464         self.record_event("terminate", event_descr)
2465
2466         # Move the state to VNF_TERMINATE_PHASE
2467         self._log.debug("Terminating VNFFGs in NS ID: %s", self.id)
2468         self.set_state(NetworkServiceRecordState.VNFFG_TERMINATE_PHASE)
2469         event_descr = "Terminating VNFFGS in NS Id:%s" % self.id
2470         self.record_event("terminating-vnffgss", event_descr)
2471         yield from terminate_vnffgrs()
2472
2473         # Move the state to VNF_TERMINATE_PHASE
2474         self.set_state(NetworkServiceRecordState.VNF_TERMINATE_PHASE)
2475         event_descr = "Terminating VNFS in NS Id:%s" % self.id
2476         self.record_event("terminating-vnfs", event_descr)
2477         yield from self.terminate_vnfrs(self.vnfrs.values())
2478
2479         # Move the state to VL_TERMINATE_PHASE
2480         self.set_state(NetworkServiceRecordState.VL_TERMINATE_PHASE)
2481         event_descr = "Terminating VLs in NS Id:%s" % self.id
2482         self.record_event("terminating-vls", event_descr)
2483         yield from terminate_vlrs()
2484
2485         yield from self.nsm_plugin.terminate_ns(self)
2486
2487         # Move the state to TERMINATED
2488         self.set_state(NetworkServiceRecordState.TERMINATED)
2489         event_descr = "Terminated NS Id:%s" % self.id
2490         self.record_event("terminated", event_descr)
2491
2492     def enable(self):
2493         """"Enable a NetworkServiceRecord."""
2494         pass
2495
2496     def disable(self):
2497         """"Disable a NetworkServiceRecord."""
2498         pass
2499
2500     def map_config_status(self):
2501         self._log.debug("Config status for ns {} is {}".
2502                         format(self.name, self._config_status))
2503         if self._config_status == NsrYang.ConfigStates.CONFIGURING:
2504             return 'configuring'
2505         if self._config_status == NsrYang.ConfigStates.FAILED:
2506             return 'failed'
2507         return 'configured'
2508
2509     def vl_phase_completed(self):
2510         """ Are VLs created in this NS?"""
2511         return self._vl_phase_completed
2512
2513     def vnf_phase_completed(self):
2514         """ Are VLs created in this NS?"""
2515         return self._vnf_phase_completed
2516
2517     def create_msg(self):
2518         """ The network serice record as a message """
2519         nsr_dict = {"ns_instance_config_ref": self.id}
2520         nsr = RwNsrYang.YangData_RwProject_Project_NsInstanceOpdata_Nsr.from_dict(nsr_dict)
2521         #nsr.cloud_account = self.cloud_account_name
2522         nsr.sdn_account = self._sdn_account_name
2523         nsr.name_ref = self.name
2524         nsr.nsd_ref = self.nsd_id
2525         nsr.nsd_name_ref = self.nsd_msg.name
2526         nsr.operational_events = self._op_status.msg
2527         nsr.operational_status = self._op_status.yang_str()
2528         nsr.config_status = self.map_config_status()
2529         nsr.config_status_details = self._config_status_details
2530         nsr.create_time = self._create_time
2531         nsr.uptime = int(time.time()) - self._create_time
2532
2533         for cfg_prim in self.nsd_msg.service_primitive:
2534             cfg_prim = NsrYang.YangData_RwProject_Project_NsInstanceOpdata_Nsr_ServicePrimitive.from_dict(
2535                     cfg_prim.as_dict())
2536             nsr.service_primitive.append(cfg_prim)
2537
2538         for init_cfg in self.nsd_msg.initial_config_primitive:
2539             prim = NsrYang.NsrInitialConfigPrimitive.from_dict(
2540                 init_cfg.as_dict())
2541             nsr.initial_config_primitive.append(prim)
2542
2543         if self.vl_phase_completed():
2544             for vlr in self.vlrs:
2545                 nsr.vlr.append(vlr.create_nsr_vlr_msg(self.vnfrs.values()))
2546
2547         if self.vnf_phase_completed():
2548             for vnfr_id in self.vnfrs:
2549                 nsr.constituent_vnfr_ref.append(self.vnfrs[vnfr_id].const_vnfr_msg)
2550             for vnffgr in self.vnffgrs.values():
2551                 nsr.vnffgr.append(vnffgr.fetch_vnffgr())
2552             for scaling_group in self._scaling_groups.values():
2553                 nsr.scaling_group_record.append(scaling_group.create_record_msg())
2554
2555         return nsr
2556
2557     def all_vnfs_active(self):
2558         """ Are all VNFS in this NS active? """
2559         for _, vnfr in self.vnfrs.items():
2560             if vnfr.active is not True:
2561                 return False
2562         return True
2563
2564     @asyncio.coroutine
2565     def update_state(self):
2566         """ Re-evaluate this  NS's state """
2567         curr_state = self._op_status.state
2568
2569         if curr_state == NetworkServiceRecordState.TERMINATED:
2570             self._log.debug("NS (%s) in terminated state, not updating state", self.id)
2571             return
2572
2573         new_state = NetworkServiceRecordState.RUNNING
2574         self._log.info("Received update_state for nsr: %s, curr-state: %s",
2575                        self.id, curr_state)
2576
2577         # Check all the VNFRs are present
2578         for _, vnfr in self.vnfrs.items():
2579             if vnfr.state in [VnfRecordState.ACTIVE, VnfRecordState.TERMINATED]:
2580                 pass
2581             elif vnfr.state == VnfRecordState.FAILED:
2582                 if vnfr._prev_state != vnfr.state:
2583                     event_descr = "Instantiation of VNF %s failed" % vnfr.id
2584                     event_error_details = vnfr.state_failed_reason
2585                     self.record_event("vnf-failed", event_descr, evt_details=event_error_details)
2586                     vnfr.set_state(VnfRecordState.FAILED)
2587                 else:
2588                     self._log.info("VNF state did not change, curr=%s, prev=%s",
2589                                    vnfr.state, vnfr._prev_state)
2590                 new_state = NetworkServiceRecordState.FAILED
2591                 break
2592             else:
2593                 self._log.info("VNF %s in NSR %s is still not active; current state is: %s",
2594                                vnfr.id, self.id, vnfr.state)
2595                 new_state = curr_state
2596
2597         # If new state is RUNNING; check all VLs
2598         if new_state == NetworkServiceRecordState.RUNNING:
2599             for vl in self.vlrs:
2600
2601                 if vl.state in [VlRecordState.ACTIVE, VlRecordState.TERMINATED]:
2602                     pass
2603                 elif vl.state == VlRecordState.FAILED:
2604                     if vl.prev_state != vl.state:
2605                         event_descr = "Instantiation of VL %s failed" % vl.id
2606                         event_error_details = vl.state_failed_reason
2607                         self.record_event("vl-failed", event_descr, evt_details=event_error_details)
2608                         vl.prev_state = vl.state
2609                     else:
2610                         self._log.debug("VL %s already in failed state")
2611                 else:
2612                     if vl.state in [VlRecordState.INSTANTIATION_PENDING, VlRecordState.INIT]:
2613                         new_state = NetworkServiceRecordState.VL_INSTANTIATE
2614                         break
2615
2616                     if vl.state in [VlRecordState.TERMINATE_PENDING]:
2617                         new_state = NetworkServiceRecordState.VL_TERMINATE
2618                         break
2619
2620         # If new state is RUNNING; check VNFFGRs are also active
2621         if new_state == NetworkServiceRecordState.RUNNING:
2622             for _, vnffgr in self.vnffgrs.items():
2623                 self._log.info("Checking vnffgr state for nsr %s is: %s",
2624                                self.id, vnffgr.state)
2625                 if vnffgr.state == VnffgRecordState.ACTIVE:
2626                     pass
2627                 elif vnffgr.state == VnffgRecordState.FAILED:
2628                     event_descr = "Instantiation of VNFFGR %s failed" % vnffgr.id
2629                     self.record_event("vnffg-failed", event_descr)
2630                     new_state = NetworkServiceRecordState.FAILED
2631                     break
2632                 else:
2633                     self._log.info("VNFFGR %s in NSR %s is still not active; current state is: %s",
2634                                     vnffgr.id, self.id, vnffgr.state)
2635                     new_state = curr_state
2636
2637         # Update all the scaling group instance operational status to
2638         # reflect the state of all VNFR within that instance
2639         yield from self._update_scale_group_instances_status()
2640
2641         for _, group in self._scaling_groups.items():
2642             if group.state == scale_group.ScaleGroupState.SCALING_OUT:
2643                 new_state = NetworkServiceRecordState.SCALING_OUT
2644                 break
2645             elif group.state == scale_group.ScaleGroupState.SCALING_IN:
2646                 new_state = NetworkServiceRecordState.SCALING_IN
2647                 break
2648
2649         if new_state != curr_state:
2650             self._log.debug("Changing state of Network service %s from %s to %s",
2651                             self.id, curr_state, new_state)
2652             if new_state == NetworkServiceRecordState.RUNNING:
2653                 yield from self.is_active()
2654             elif new_state == NetworkServiceRecordState.FAILED:
2655                 # If the NS is already active and we entered scaling_in, scaling_out,
2656                 # do not mark the NS as failing if scaling operation failed.
2657                 if curr_state in [NetworkServiceRecordState.SCALING_OUT,
2658                                   NetworkServiceRecordState.SCALING_IN] and self._is_active:
2659                     new_state = NetworkServiceRecordState.RUNNING
2660                     self.set_state(new_state)
2661                 else:
2662                     yield from self.instantiation_failed()
2663             else:
2664                 self.set_state(new_state)
2665
2666         yield from self.publish()
2667
2668
2669 class InputParameterSubstitution(object):
2670     """
2671     This class is responsible for substituting input parameters into an NSD.
2672     """
2673
2674     def __init__(self, log, project):
2675         """Create an instance of InputParameterSubstitution
2676
2677         Arguments:
2678             log - a logger for this object to use
2679
2680         """
2681         self.log = log
2682         self.project = project
2683
2684     def __call__(self, nsd, nsr_config):
2685         """Substitutes input parameters from the NSR config into the NSD
2686
2687         This call modifies the provided NSD with the input parameters that are
2688         contained in the NSR config.
2689
2690         Arguments:
2691             nsd        - a GI NSD object
2692             nsr_config - a GI NSR config object
2693
2694         """
2695         if nsd is None or nsr_config is None:
2696             return
2697
2698         # Create a lookup of the xpath elements that this descriptor allows
2699         # to be modified
2700         optional_input_parameters = set()
2701         for input_parameter in nsd.input_parameter_xpath:
2702             optional_input_parameters.add(self.project.add_project(input_parameter.xpath))
2703
2704         # Apply the input parameters to the descriptor
2705         if nsr_config.input_parameter:
2706             for param in nsr_config.input_parameter:
2707                 if param.xpath not in optional_input_parameters:
2708                     msg = "tried to set an invalid input parameter ({})"
2709                     self.log.error(msg.format(param.xpath))
2710                     continue
2711
2712                 self.log.debug(
2713                         "input-parameter:{} = {}".format(
2714                             param.xpath,
2715                             param.value,
2716                             )
2717                         )
2718
2719                 try:
2720                     xpath.setxattr(nsd, param.xpath, param.value)
2721
2722                 except Exception as e:
2723                     self.log.exception(e)
2724
2725
2726 class NetworkServiceDescriptor(object):
2727     """
2728     Network service descriptor class
2729     """
2730
2731     def __init__(self, dts, log, loop, nsd, nsm):
2732         self._dts = dts
2733         self._log = log
2734         self._loop = loop
2735
2736         self._nsd = nsd
2737         self._ref_count = 0
2738
2739         self._nsm = nsm
2740
2741     @property
2742     def id(self):
2743         """ Returns nsd id """
2744         return self._nsd.id
2745
2746     @property
2747     def name(self):
2748         """ Returns name of nsd """
2749         return self._nsd.name
2750
2751     @property
2752     def ref_count(self):
2753         """ Returns reference count"""
2754         return self._ref_count
2755
2756     def in_use(self):
2757         """ Returns whether nsd is in use or not """
2758         return True if self.ref_count > 0 else False
2759
2760     def ref(self):
2761         """ Take a reference on this object """
2762         self._ref_count += 1
2763
2764     def unref(self):
2765         """ Release reference on this object """
2766         if self.ref_count < 1:
2767             msg = ("Unref on a NSD object - nsd id %s, ref_count = %s" %
2768                    (self.id, self.ref_count))
2769             self._log.critical(msg)
2770             raise NetworkServiceDescriptorError(msg)
2771         self._ref_count -= 1
2772
2773     @property
2774     def msg(self):
2775         """ Return the message associated with this NetworkServiceDescriptor"""
2776         return self._nsd
2777
2778     @staticmethod
2779     def path_for_id(nsd_id):
2780         """ Return path for the passed nsd_id"""
2781         return self._nsm._project.add_project(
2782             "C,/project-nsd:nsd-catalog/project-nsd:nsd[project-nsd:id = '{}'".
2783             format(nsd_id))
2784
2785     def path(self):
2786         """ Return the message associated with this NetworkServiceDescriptor"""
2787         return NetworkServiceDescriptor.path_for_id(self.id)
2788
2789     def update(self, nsd):
2790         """ Update the NSD descriptor """
2791         self._nsd = nsd
2792
2793
2794 class NsdDtsHandler(object):
2795     """ The network service descriptor DTS handler """
2796     XPATH = "C,/project-nsd:nsd-catalog/project-nsd:nsd"
2797
2798     def __init__(self, dts, log, loop, nsm):
2799         self._dts = dts
2800         self._log = log
2801         self._loop = loop
2802         self._nsm = nsm
2803
2804         self._regh = None
2805         self._project = nsm._project
2806
2807     @property
2808     def regh(self):
2809         """ Return registration handle """
2810         return self._regh
2811
2812     @asyncio.coroutine
2813     def register(self):
2814         """ Register for Nsd create/update/delete/read requests from dts """
2815
2816         if self._regh:
2817             self._log.warning("DTS handler already registered for project {}".
2818                               format(self._project.name))
2819             return
2820
2821         def on_apply(dts, acg, xact, action, scratch):
2822             """Apply the  configuration"""
2823             is_recovery = xact.xact is None and action == rwdts.AppconfAction.INSTALL
2824             self._log.debug("Got nsd apply cfg (xact:%s) (action:%s)",
2825                             xact, action)
2826
2827             if self._regh:
2828                 # Create/Update an NSD record
2829                 for cfg in self._regh.get_xact_elements(xact):
2830                     # Only interested in those NSD cfgs whose ID was received in prepare callback
2831                     if cfg.id in scratch.get('nsds', []) or is_recovery:
2832                         self._nsm.update_nsd(cfg)
2833
2834             else:
2835                 self._log.error("No reg handle for {} for project {}".
2836                                 format(self.__class__, self._project.name))
2837
2838             scratch.pop('nsds', None)
2839
2840             return RwTypes.RwStatus.SUCCESS
2841
2842         @asyncio.coroutine
2843         def delete_nsd_libs(nsd_id):
2844             """ Remove any files uploaded with NSD and stored under $RIFT_ARTIFACTS/libs/<id> """
2845             try:
2846                 rift_artifacts_dir = os.environ['RIFT_ARTIFACTS']
2847                 nsd_dir = os.path.join(rift_artifacts_dir, 'launchpad/libs', nsd_id)
2848
2849                 if os.path.exists (nsd_dir):
2850                     shutil.rmtree(nsd_dir, ignore_errors=True)
2851             except Exception as e:
2852                 self._log.error("Exception in cleaning up NSD libs {}: {}".
2853                                 format(nsd_id, e))
2854                 self._log.exception(e)
2855
2856         @asyncio.coroutine
2857         def on_prepare(dts, acg, xact, xact_info, ks_path, msg, scratch):
2858             """ Prepare callback from DTS for NSD config """
2859
2860             self._log.info("Got nsd prepare - config received nsd id %s, msg %s",
2861                            msg.id, msg)
2862
2863             fref = ProtobufC.FieldReference.alloc()
2864             fref.goto_whole_message(msg.to_pbcm())
2865
2866             if fref.is_field_deleted():
2867                 # Delete an NSD record
2868                 self._log.debug("Deleting NSD with id %s", msg.id)
2869                 if self._nsm.nsd_in_use(msg.id):
2870                     self._log.debug("Cannot delete NSD in use - %s", msg.id)
2871                     err = "Cannot delete an NSD in use - %s" % msg.id
2872                     raise NetworkServiceDescriptorRefCountExists(err)
2873
2874                 yield from delete_nsd_libs(msg.id)
2875                 self._nsm.delete_nsd(msg.id)
2876             else:
2877                 # Add this NSD to scratch to create/update in apply callback
2878                 nsds = scratch.setdefault('nsds', [])
2879                 nsds.append(msg.id)
2880                 # acg._scratch['nsds'].append(msg.id)
2881
2882             xact_info.respond_xpath(rwdts.XactRspCode.ACK)
2883
2884         self._log.debug(
2885             "Registering for NSD config using xpath: %s",
2886             NsdDtsHandler.XPATH,
2887             )
2888
2889         acg_hdl = rift.tasklets.AppConfGroup.Handler(on_apply=on_apply)
2890         with self._dts.appconf_group_create(handler=acg_hdl) as acg:
2891             # Need a list in scratch to store NSDs to create/update later
2892             # acg._scratch['nsds'] = list()
2893             self._regh = acg.register(
2894                 xpath=self._project.add_project(NsdDtsHandler.XPATH),
2895                 flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY | rwdts.Flag.CACHE,
2896                 on_prepare=on_prepare)
2897
2898     def deregister(self):
2899         self._log.debug("De-register NSD handler for project {}".
2900                         format(self._project.name))
2901         if self._regh:
2902             self._regh.deregister()
2903             self._regh = None
2904
2905
2906 class VnfdDtsHandler(object):
2907     """ DTS handler for VNFD config changes """
2908     XPATH = "C,/project-vnfd:vnfd-catalog/project-vnfd:vnfd"
2909
2910     def __init__(self, dts, log, loop, nsm):
2911         self._dts = dts
2912         self._log = log
2913         self._loop = loop
2914         self._nsm = nsm
2915         self._regh = None
2916         self._project = nsm._project
2917
2918     @property
2919     def regh(self):
2920         """ DTS registration handle """
2921         return self._regh
2922
2923     @asyncio.coroutine
2924     def register(self):
2925         """ Register for VNFD configuration"""
2926
2927         if self._regh:
2928             self._log.warning("DTS handler already registered for project {}".
2929                               format(self._project.name))
2930             return
2931
2932         @asyncio.coroutine
2933         def on_apply(dts, acg, xact, action, scratch):
2934             """Apply the  configuration"""
2935             self._log.debug("Got NSM VNFD apply (xact: %s) (action: %s)(scr: %s)",
2936                             xact, action, scratch)
2937
2938             if self._regh:
2939                 # Create/Update a VNFD record
2940                 for cfg in self._regh.get_xact_elements(xact):
2941                     # Only interested in those VNFD cfgs whose ID was received in prepare callback
2942                     if cfg.id in scratch.get('vnfds', []):
2943                         self._nsm.update_vnfd(cfg)
2944
2945                         for cfg in self._regh.elements:
2946                             if cfg.id in scratch.get('deleted_vnfds', []):
2947                                 yield from self._nsm.delete_vnfd(cfg.id)
2948
2949             else:
2950                 self._log.error("Reg handle none for {} in project {}".
2951                                 format(self.__class__, self._project))
2952
2953             scratch.pop('vnfds', None)
2954             scratch.pop('deleted_vnfds', None)
2955
2956         @asyncio.coroutine
2957         def on_prepare(dts, acg, xact, xact_info, ks_path, msg, scratch):
2958             """ on prepare callback """
2959             self._log.debug("Got on prepare for VNFD (path: %s) (action: %s) (msg: %s)",
2960                             ks_path.to_xpath(RwNsmYang.get_schema()), xact_info.query_action, msg)
2961
2962             fref = ProtobufC.FieldReference.alloc()
2963             fref.goto_whole_message(msg.to_pbcm())
2964
2965             # Handle deletes in prepare_callback, but adds/updates in apply_callback
2966             if fref.is_field_deleted():
2967                 self._log.debug("Adding msg to deleted field")
2968                 deleted_vnfds = scratch.setdefault('deleted_vnfds', [])
2969                 deleted_vnfds.append(msg.id)
2970             else:
2971                 # Add this VNFD to scratch to create/update in apply callback
2972                 vnfds = scratch.setdefault('vnfds', [])
2973                 vnfds.append(msg.id)
2974
2975             xact_info.respond_xpath(rwdts.XactRspCode.ACK)
2976
2977         xpath = self._project.add_project(VnfdDtsHandler.XPATH)
2978         self._log.debug(
2979             "Registering for VNFD config using xpath {} for project {}"
2980             .format(xpath, self._project))
2981         acg_hdl = rift.tasklets.AppConfGroup.Handler(on_apply=on_apply)
2982         with self._dts.appconf_group_create(handler=acg_hdl) as acg:
2983             # Need a list in scratch to store VNFDs to create/update later
2984             # acg._scratch['vnfds'] = list()
2985             # acg._scratch['deleted_vnfds'] = list()
2986             self._regh = acg.register(
2987                 xpath=xpath,
2988                 flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY,
2989                 on_prepare=on_prepare)
2990
2991     def deregister(self):
2992         self._log.debug("De-register VNFD handler for project {}".
2993                         format(self._project.name))
2994         if self._regh:
2995             self._regh.deregister()
2996             self._regh = None
2997
2998
2999 class NsrRpcDtsHandler(object):
3000     """ The network service instantiation RPC DTS handler """
3001     EXEC_NSR_CONF_XPATH = "I,/nsr:start-network-service"
3002     EXEC_NSR_CONF_O_XPATH = "O,/nsr:start-network-service"
3003     NETCONF_IP_ADDRESS = "127.0.0.1"
3004     NETCONF_PORT = 2022
3005     RESTCONF_PORT = 8888
3006     NETCONF_USER = "admin"
3007     NETCONF_PW = "admin"
3008     REST_BASE_V2_URL = 'https://{}:{}/v2/api/'.format("127.0.0.1",8888)
3009
3010     def __init__(self, dts, log, loop, nsm):
3011         self._dts = dts
3012         self._log = log
3013         self._loop = loop
3014         self._nsm = nsm
3015         self._nsd = None
3016
3017         self._ns_regh = None
3018
3019         self._manager = None
3020         self._nsr_config_url = NsrRpcDtsHandler.REST_BASE_V2_URL + \
3021                                'config/project/{}/ns-instance-config'. \
3022                                format(self._nsm._project.name)
3023
3024         self._model = RwYang.Model.create_libncx()
3025         self._model.load_schema_ypbc(RwNsrYang.get_schema())
3026
3027     @property
3028     def nsm(self):
3029         """ Return the NS manager instance """
3030         return self._nsm
3031
3032     @staticmethod
3033     def wrap_netconf_config_xml(xml):
3034         xml = '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml)
3035         return xml
3036
3037     @asyncio.coroutine
3038     def _connect(self, timeout_secs=240):
3039
3040         start_time = time.time()
3041         while (time.time() - start_time) < timeout_secs:
3042
3043             try:
3044                 self._log.debug("Attemping NsmTasklet netconf connection.")
3045
3046                 manager = yield from ncclient.asyncio_manager.asyncio_connect(
3047                         loop=self._loop,
3048                         host=NsrRpcDtsHandler.NETCONF_IP_ADDRESS,
3049                         port=NsrRpcDtsHandler.NETCONF_PORT,
3050                         username=NsrRpcDtsHandler.NETCONF_USER,
3051                         password=NsrRpcDtsHandler.NETCONF_PW,
3052                         allow_agent=False,
3053                         look_for_keys=False,
3054                         hostkey_verify=False,
3055                         )
3056
3057                 return manager
3058
3059             except ncclient.transport.errors.SSHError as e:
3060                 self._log.warning("Netconf connection to launchpad %s failed: %s",
3061                                   NsrRpcDtsHandler.NETCONF_IP_ADDRESS, str(e))
3062
3063             yield from asyncio.sleep(5, loop=self._loop)
3064
3065         raise NsrInstantiationFailed("Failed to connect to Launchpad within %s seconds" %
3066                                       timeout_secs)
3067
3068     def _apply_ns_instance_config(self,payload_dict):
3069         #self._log.debug("At apply NS instance config with payload %s",payload_dict)
3070         req_hdr= {'accept':'application/vnd.yang.data+json',
3071                   'content-type':'application/vnd.yang.data+json'}
3072         response=requests.post(self._nsr_config_url, headers=req_hdr,
3073                                auth=('admin', 'admin'),data=payload_dict,verify=False)
3074         return response
3075
3076     @asyncio.coroutine
3077     def register(self):
3078         """ Register for NS monitoring read from dts """
3079         if self._ns_regh:
3080             self._log.warning("RPC already registered for project {}".
3081                               format(self._project.name))
3082             return
3083
3084         @asyncio.coroutine
3085         def on_ns_config_prepare(xact_info, action, ks_path, msg):
3086             """ prepare callback from dts start-network-service"""
3087             assert action == rwdts.QueryAction.RPC
3088             rpc_ip = msg
3089
3090             if not self._nsm._project.rpc_check(msg, xact_info=xact_info):
3091                 return
3092
3093             rpc_op = NsrYang.YangOutput_Nsr_StartNetworkService.from_dict({
3094                 "nsr_id":str(uuid.uuid4()),
3095                 "project_name": msg.prject_name,
3096             })
3097
3098             if not ('name' in rpc_ip and  'nsd_ref' in rpc_ip and
3099                     ('cloud_account' in rpc_ip or 'om_datacenter' in rpc_ip)):
3100                 self._log.error("Mandatory parameters name or nsd_ref or " +
3101                                 "cloud account not found in start-network-service {}".
3102                                 format(rpc_ip))
3103
3104
3105             self._log.debug("start-network-service RPC input: {}".format(rpc_ip))
3106
3107             try:
3108                 # Add used value to the pool
3109                 self._log.debug("RPC output: {}".format(rpc_op))
3110
3111                 nsd_copy = self.nsm.get_nsd(rpc_ip.nsd_ref)
3112
3113                 #if not self._manager:
3114                 #    self._manager = yield from self._connect()
3115
3116                 self._log.debug("Configuring ns-instance-config with name  %s nsd-ref: %s",
3117                         rpc_ip.name, rpc_ip.nsd_ref)
3118
3119                 ns_instance_config_dict = {"id":rpc_op.nsr_id, "admin_status":"ENABLED"}
3120                 ns_instance_config_copy_dict = {k:v for k, v in rpc_ip.as_dict().items()
3121                                                 if k in RwNsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr().fields}
3122                 ns_instance_config_dict.update(ns_instance_config_copy_dict)
3123
3124                 ns_instance_config = RwNsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr.from_dict(ns_instance_config_dict)
3125                 ns_instance_config.nsd = NsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr_Nsd()
3126                 ns_instance_config.nsd.from_dict(nsd_copy.msg.as_dict())
3127
3128                 payload_dict = ns_instance_config.to_json(self._model)
3129                 #xml = ns_instance_config.to_xml_v2(self._model)
3130                 #netconf_xml = self.wrap_netconf_config_xml(xml)
3131
3132                 #self._log.debug("Sending configure ns-instance-config xml to %s: %s",
3133                 #        netconf_xml, NsrRpcDtsHandler.NETCONF_IP_ADDRESS)
3134                 self._log.debug("Sending configure ns-instance-config json to %s: %s",
3135                         self._nsr_config_url,ns_instance_config)
3136
3137                 #response = yield from self._manager.edit_config(
3138                 #           target="running",
3139                 #           config=netconf_xml,
3140                 #           )
3141                 response = yield from self._loop.run_in_executor(
3142                     None,
3143                     self._apply_ns_instance_config,
3144                     payload_dict
3145                     )
3146                 response.raise_for_status()
3147                 self._log.debug("Received edit config response: %s", response.json())
3148
3149                 xact_info.respond_xpath(rwdts.XactRspCode.ACK,
3150                                         NsrRpcDtsHandler.EXEC_NSR_CONF_O_XPATH,
3151                                         rpc_op)
3152             except Exception as e:
3153                 self._log.error("Exception processing the "
3154                                 "start-network-service: {}".format(e))
3155                 self._log.exception(e)
3156                 xact_info.respond_xpath(rwdts.XactRspCode.NACK,
3157                                         NsrRpcDtsHandler.EXEC_NSR_CONF_O_XPATH)
3158
3159
3160         hdl_ns = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_ns_config_prepare,)
3161
3162         with self._dts.group_create() as group:
3163             self._ns_regh = group.register(xpath=NsrRpcDtsHandler.EXEC_NSR_CONF_XPATH,
3164                                            handler=hdl_ns,
3165                                            flags=rwdts.Flag.PUBLISHER,
3166                                           )
3167
3168     def deregister(self):
3169         self._log.debug("De-register NSR RPC for project {}".
3170                         format(self._nsm._project.name))
3171         if self._ns_regh:
3172             self._ns_regh.deregister()
3173             self._ns_regh = None
3174
3175
3176 class NsrDtsHandler(object):
3177     """ The network service DTS handler """
3178     NSR_XPATH = "C,/nsr:ns-instance-config/nsr:nsr"
3179     SCALE_INSTANCE_XPATH = "C,/nsr:ns-instance-config/nsr:nsr/nsr:scaling-group/nsr:instance"
3180     KEY_PAIR_XPATH = "C,/nsr:key-pair"
3181
3182     def __init__(self, dts, log, loop, nsm):
3183         self._dts = dts
3184         self._log = log
3185         self._loop = loop
3186         self._nsm = nsm
3187         self._project = self._nsm._project
3188
3189         self._nsr_regh = None
3190         self._scale_regh = None
3191         self._key_pair_regh = None
3192
3193     @property
3194     def nsm(self):
3195         """ Return the NS manager instance """
3196         return self._nsm
3197
3198     @asyncio.coroutine
3199     def register(self):
3200         """ Register for Nsr create/update/delete/read requests from dts """
3201
3202         if self._nsr_regh:
3203             self._log.warning("DTS handler already registered for project {}".
3204                               format(self._project.name))
3205             return
3206
3207         def nsr_id_from_keyspec(ks):
3208             nsr_path_entry = NsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr.schema().keyspec_to_entry(ks)
3209             nsr_id = nsr_path_entry.key00.id
3210             return nsr_id
3211
3212         def group_name_from_keyspec(ks):
3213             group_path_entry = NsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup.schema().keyspec_to_entry(ks)
3214             group_name = group_path_entry.key00.scaling_group_name_ref
3215             return group_name
3216
3217         def is_instance_in_reg_elements(nsr_id, group_name, instance_id):
3218             """ Return boolean indicating if scaling group instance was already commited previously.
3219
3220             By looking at the existing elements in this registration handle (elements not part
3221             of this current xact), we can tell if the instance was configured previously without
3222             keeping any application state.
3223             """
3224             for instance_cfg, keyspec in self._nsr_regh.get_xact_elements(include_keyspec=True):
3225                 elem_nsr_id = nsr_id_from_keyspec(keyspec)
3226                 elem_group_name = group_name_from_keyspec(keyspec)
3227
3228                 if elem_nsr_id != nsr_id or group_name != elem_group_name:
3229                     continue
3230
3231                 if instance_cfg.id == instance_id:
3232                     return True
3233
3234             return False
3235
3236         def get_scale_group_instance_delta(nsr_id, group_name, xact):
3237             delta = {"added": [], "deleted": []}
3238             for instance_cfg, keyspec in self._scale_regh.get_xact_elements(xact, include_keyspec=True):
3239                 elem_nsr_id = nsr_id_from_keyspec(keyspec)
3240                 if elem_nsr_id != nsr_id:
3241                     continue
3242
3243                 elem_group_name = group_name_from_keyspec(keyspec)
3244                 if elem_group_name != group_name:
3245                     continue
3246
3247                 delta["added"].append(instance_cfg.id)
3248
3249             for instance_cfg, keyspec in self._scale_regh.get_xact_elements(include_keyspec=True):
3250                 elem_nsr_id = nsr_id_from_keyspec(keyspec)
3251                 if elem_nsr_id != nsr_id:
3252                     continue
3253
3254                 elem_group_name = group_name_from_keyspec(keyspec)
3255                 if elem_group_name != group_name:
3256                     continue
3257
3258                 if instance_cfg.id in delta["added"]:
3259                     delta["added"].remove(instance_cfg.id)
3260                 else:
3261                     delta["deleted"].append(instance_cfg.id)
3262
3263             return delta
3264
3265         @asyncio.coroutine
3266         def update_nsr_nsd(nsr_id, xact, scratch):
3267
3268             @asyncio.coroutine
3269             def get_nsr_vl_delta(nsr_id, xact, scratch):
3270                 delta = {"added": [], "deleted": []}
3271                 for instance_cfg, keyspec in self._nsr_regh.get_xact_elements(xact, include_keyspec=True):
3272                     elem_nsr_id = nsr_id_from_keyspec(keyspec)
3273                     if elem_nsr_id != nsr_id:
3274                         continue
3275
3276                     if 'vld' in instance_cfg.nsd:
3277                         for vld in instance_cfg.nsd.vld:
3278                             delta["added"].append(vld)
3279
3280                 for instance_cfg, keyspec in self._nsr_regh.get_xact_elements(include_keyspec=True):
3281                     self._log.debug("NSR update: %s", instance_cfg)
3282                     elem_nsr_id = nsr_id_from_keyspec(keyspec)
3283                     if elem_nsr_id != nsr_id:
3284                         continue
3285
3286                     if 'vld' in instance_cfg.nsd:
3287                         for vld in instance_cfg.nsd.vld:
3288                             if vld in delta["added"]:
3289                                 delta["added"].remove(vld)
3290                             else:
3291                                 delta["deleted"].append(vld)
3292
3293                 return delta
3294
3295             vl_delta = yield from get_nsr_vl_delta(nsr_id, xact, scratch)
3296             self._log.debug("Got NSR:%s VL instance delta: %s", nsr_id, vl_delta)
3297
3298             for vld in vl_delta["added"]:
3299                 yield from self._nsm.nsr_instantiate_vl(nsr_id, vld)
3300
3301             for vld in vl_delta["deleted"]:
3302                 yield from self._nsm.nsr_terminate_vl(nsr_id, vld)
3303
3304         def get_nsr_key_pairs(dts_member_reg, xact):
3305             key_pairs = {}
3306             for instance_cfg, keyspec in dts_member_reg.get_xact_elements(xact, include_keyspec=True):
3307                 self._log.debug("Key pair received is {} KS: {}".format(instance_cfg, keyspec))
3308                 xpath = keyspec.to_xpath(RwNsrYang.get_schema())
3309                 key_pairs[instance_cfg.name] = instance_cfg
3310             return key_pairs
3311
3312         def on_apply(dts, acg, xact, action, scratch):
3313             """Apply the  configuration"""
3314             self._log.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3315                             xact, action, scratch)
3316
3317             def handle_create_nsr(msg, key_pairs=None, restart_mode=False):
3318                 # Handle create nsr requests """
3319                 # Do some validations
3320                 if not msg.has_field("nsd"):
3321                     err = "NSD not provided"
3322                     self._log.error(err)
3323                     raise NetworkServiceRecordError(err)
3324
3325                 self._log.debug("Creating NetworkServiceRecord %s  from nsr config  %s",
3326                                msg.id, msg.as_dict())
3327                 nsr = self.nsm.create_nsr(msg, key_pairs=key_pairs, restart_mode=restart_mode)
3328                 return nsr
3329
3330             def handle_delete_nsr(msg):
3331                 @asyncio.coroutine
3332                 def delete_instantiation(ns_id):
3333                     """ Delete instantiation """
3334                     with self._dts.transaction() as xact:
3335                         yield from self._nsm.terminate_ns(ns_id, xact)
3336
3337                 # Handle delete NSR requests
3338                 self._log.info("Delete req for  NSR Id: %s received", msg.id)
3339                 # Terminate the NSR instance
3340                 nsr = self._nsm.get_ns_by_nsr_id(msg.id)
3341
3342                 nsr.set_state(NetworkServiceRecordState.TERMINATE_RCVD)
3343                 event_descr = "Terminate rcvd for NS Id:%s" % msg.id
3344                 nsr.record_event("terminate-rcvd", event_descr)
3345
3346                 self._loop.create_task(delete_instantiation(msg.id))
3347
3348             @asyncio.coroutine
3349             def begin_instantiation(nsr):
3350                 # Begin instantiation
3351                 self._log.info("Beginning NS instantiation: %s", nsr.id)
3352                 try:
3353                     yield from self._nsm.instantiate_ns(nsr.id, xact)
3354                 except Exception as e:
3355                     self._log.exception("NS instantiation: {}".format(e))
3356                     raise e
3357
3358             self._log.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3359                             xact, action, scratch)
3360
3361             if action == rwdts.AppconfAction.INSTALL and xact.id is None:
3362                 key_pairs = []
3363                 if self._key_pair_regh:
3364                     for element in self._key_pair_regh.elements:
3365                         key_pairs.append(element)
3366                 else:
3367                     self._log.error("Reg handle none for key pair in project {}".
3368                                     format(self._project))
3369
3370                 if self._nsr_regh:
3371                     for element in self._nsr_regh.elements:
3372                         nsr = handle_create_nsr(element, key_pairs, restart_mode=True)
3373                         self._loop.create_task(begin_instantiation(nsr))
3374                 else:
3375                     self._log.error("Reg handle none for NSR in project {}".
3376                                     format(self._project))
3377
3378
3379             (added_msgs, deleted_msgs, updated_msgs) = get_add_delete_update_cfgs(self._nsr_regh,
3380                                                                                   xact,
3381                                                                                   "id")
3382             self._log.debug("Added: %s, Deleted: %s, Updated: %s", added_msgs,
3383                             deleted_msgs, updated_msgs)
3384
3385             for msg in added_msgs:
3386                 if msg.id not in self._nsm.nsrs:
3387                     self._log.info("Create NSR received in on_apply to instantiate NS:%s", msg.id)
3388                     key_pairs = get_nsr_key_pairs(self._key_pair_regh, xact)
3389                     nsr = handle_create_nsr(msg,key_pairs)
3390                     self._loop.create_task(begin_instantiation(nsr))
3391
3392             for msg in deleted_msgs:
3393                 self._log.info("Delete NSR received in on_apply to terminate NS:%s", msg.id)
3394                 try:
3395                     handle_delete_nsr(msg)
3396                 except Exception:
3397                     self._log.exception("Failed to terminate NS:%s", msg.id)
3398
3399             for msg in updated_msgs:
3400                 self._log.info("Update NSR received in on_apply: %s", msg)
3401
3402                 self._nsm.nsr_update_cfg(msg.id, msg)
3403
3404                 if 'nsd' in msg:
3405                     self._loop.create_task(update_nsr_nsd(msg.id, xact, scratch))
3406
3407                 for group in msg.scaling_group:
3408                     instance_delta = get_scale_group_instance_delta(msg.id, group.scaling_group_name_ref, xact)
3409                     self._log.debug("Got NSR:%s scale group instance delta: %s", msg.id, instance_delta)
3410
3411                     for instance_id in instance_delta["added"]:
3412                         self._nsm.scale_nsr_out(msg.id, group.scaling_group_name_ref, instance_id, xact)
3413
3414                     for instance_id in instance_delta["deleted"]:
3415                         self._nsm.scale_nsr_in(msg.id, group.scaling_group_name_ref, instance_id)
3416
3417
3418             return RwTypes.RwStatus.SUCCESS
3419
3420         @asyncio.coroutine
3421         def on_prepare(dts, acg, xact, xact_info, ks_path, msg, scratch):
3422             """ Prepare calllback from DTS for NSR """
3423
3424             xpath = ks_path.to_xpath(RwNsrYang.get_schema())
3425             action = xact_info.query_action
3426             self._log.debug(
3427                     "Got Nsr prepare callback (xact: %s) (action: %s) (info: %s), %s:%s)",
3428                     xact, action, xact_info, xpath, msg
3429                     )
3430
3431             @asyncio.coroutine
3432             def delete_instantiation(ns_id):
3433                 """ Delete instantiation """
3434                 yield from self._nsm.terminate_ns(ns_id, None)
3435
3436             def handle_delete_nsr():
3437                 """ Handle delete NSR requests """
3438                 self._log.info("Delete req for  NSR Id: %s received", msg.id)
3439                 # Terminate the NSR instance
3440                 nsr = self._nsm.get_ns_by_nsr_id(msg.id)
3441
3442                 nsr.set_state(NetworkServiceRecordState.TERMINATE_RCVD)
3443                 event_descr = "Terminate rcvd for NS Id:%s" % msg.id
3444                 nsr.record_event("terminate-rcvd", event_descr)
3445
3446                 self._loop.create_task(delete_instantiation(msg.id))
3447
3448             fref = ProtobufC.FieldReference.alloc()
3449             fref.goto_whole_message(msg.to_pbcm())
3450
3451             if action in [rwdts.QueryAction.CREATE, rwdts.QueryAction.UPDATE, rwdts.QueryAction.DELETE]:
3452                 # if this is an NSR create
3453                 if action != rwdts.QueryAction.DELETE and msg.id not in self._nsm.nsrs:
3454                     # Ensure the Cloud account/datacenter has been specified
3455                     if not msg.has_field("cloud_account") and not msg.has_field("om_datacenter"):
3456                         raise NsrInstantiationFailed("Cloud account or datacenter not specified in NSR")
3457
3458                     # Check if nsd is specified
3459                     if not msg.has_field("nsd"):
3460                         raise NsrInstantiationFailed("NSD not specified in NSR")
3461
3462                 else:
3463                     nsr = self._nsm.nsrs[msg.id]
3464
3465                     if msg.has_field("nsd"):
3466                         if nsr.state != NetworkServiceRecordState.RUNNING:
3467                             raise NsrVlUpdateError("Unable to update VL when NSR not in running state")
3468                         if 'vld' not in msg.nsd or len(msg.nsd.vld) == 0:
3469                             raise NsrVlUpdateError("NS config NSD should have atleast 1 VLD defined")
3470
3471                     if msg.has_field("scaling_group"):
3472                         if nsr.state != NetworkServiceRecordState.RUNNING:
3473                             raise ScalingOperationError("Unable to perform scaling action when NS is not in running state")
3474
3475                         if len(msg.scaling_group) > 1:
3476                             raise ScalingOperationError("Only a single scaling group can be configured at a time")
3477
3478                         for group_msg in msg.scaling_group:
3479                             num_new_group_instances = len(group_msg.instance)
3480                             if num_new_group_instances > 1:
3481                                 raise ScalingOperationError("Only a single scaling instance can be modified at a time")
3482
3483                             elif num_new_group_instances == 1:
3484                                 scale_group = nsr.scaling_groups[group_msg.scaling_group_name_ref]
3485                                 if action in [rwdts.QueryAction.CREATE, rwdts.QueryAction.UPDATE]:
3486                                     if len(scale_group.instances) == scale_group.max_instance_count:
3487                                         raise ScalingOperationError("Max instances for %s reached" % scale_group)
3488
3489             acg.handle.prepare_complete_ok(xact_info.handle)
3490
3491
3492         xpath = self._project.add_project(NsrDtsHandler.NSR_XPATH)
3493         self._log.debug("Registering for NSR config using xpath: {}".
3494                         format(xpath))
3495
3496         acg_hdl = rift.tasklets.AppConfGroup.Handler(on_apply=on_apply)
3497         with self._dts.appconf_group_create(handler=acg_hdl) as acg:
3498             self._nsr_regh = acg.register(
3499                 xpath=xpath,
3500                 flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY | rwdts.Flag.CACHE,
3501                 on_prepare=on_prepare
3502             )
3503
3504             self._scale_regh = acg.register(
3505                 xpath=self._project.add_project(NsrDtsHandler.SCALE_INSTANCE_XPATH),
3506                 flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY| rwdts.Flag.CACHE,
3507             )
3508
3509             self._key_pair_regh = acg.register(
3510                 xpath=self._project.add_project(NsrDtsHandler.KEY_PAIR_XPATH),
3511                 flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY | rwdts.Flag.CACHE,
3512             )
3513
3514     def deregister(self):
3515         self._log.debug("De-register NSR config for project {}".
3516                         format(self._project.name))
3517         if self._nsr_regh:
3518             self._nsr_regh.deregister()
3519             self._nsr_regh = None
3520         if self._scale_regh:
3521             self._scale_regh.deregister()
3522             self._scale_regh = None
3523         if self._key_pair_regh:
3524             self._key_pair_regh.deregister()
3525             self._key_pair_regh = None
3526
3527
3528 class NsrOpDataDtsHandler(object):
3529     """ The network service op data DTS handler """
3530     XPATH = "D,/nsr:ns-instance-opdata/nsr:nsr"
3531
3532     def __init__(self, dts, log, loop, nsm):
3533         self._dts = dts
3534         self._log = log
3535         self._loop = loop
3536         self._nsm = nsm
3537
3538         self._project = nsm._project
3539         self._regh = None
3540
3541     @property
3542     def regh(self):
3543         """ Return the registration handle"""
3544         return self._regh
3545
3546     @property
3547     def nsm(self):
3548         """ Return the NS manager instance """
3549         return self._nsm
3550
3551     @asyncio.coroutine
3552     def register(self):
3553         """ Register for Nsr op data publisher registration"""
3554         if self._regh:
3555             self._log.warning("NSR op data handler already registered for project {}".
3556                               format(self._project.name))
3557             return
3558
3559         xpath = self._project.add_project(NsrOpDataDtsHandler.XPATH)
3560         self._log.debug("Registering Nsr op data path {} as publisher".
3561                         format(xpath))
3562
3563         hdl = rift.tasklets.DTS.RegistrationHandler()
3564         handlers = rift.tasklets.Group.Handler()
3565         with self._dts.group_create(handler=handlers) as group:
3566             self._regh = group.register(xpath=xpath,
3567                                         handler=hdl,
3568                                         flags=rwdts.Flag.PUBLISHER | rwdts.Flag.NO_PREP_READ | rwdts.Flag.DATASTORE)
3569
3570     def deregister(self):
3571         self._log.debug("De-register NSR opdata for project {}".
3572                         format(self._project.name))
3573         if self._regh:
3574             self._regh.deregister()
3575             self._regh = None
3576
3577     @asyncio.coroutine
3578     def create(self, xpath, msg):
3579         """
3580         Create an NS record in DTS with the path and message
3581         """
3582         path = self._project.add_project(xpath)
3583         self._log.debug("Creating NSR %s:%s", path, msg)
3584         self.regh.create_element(path, msg)
3585         self._log.debug("Created NSR, %s:%s", path, msg)
3586
3587     @asyncio.coroutine
3588     def update(self, xpath, msg, flags=rwdts.XactFlag.REPLACE):
3589         """
3590         Update an NS record in DTS with the path and message
3591         """
3592         path = self._project.add_project(xpath)
3593         self._log.debug("Updating NSR, %s:%s regh = %s", path, msg, self.regh)
3594         self.regh.update_element(path, msg, flags)
3595         self._log.debug("Updated NSR, %s:%s", path, msg)
3596
3597     @asyncio.coroutine
3598     def delete(self, xpath):
3599         """
3600         Update an NS record in DTS with the path and message
3601         """
3602         path = self._project.add_project(xpath)
3603         self._log.debug("Deleting NSR path:%s", path)
3604         self.regh.delete_element(path)
3605         self._log.debug("Deleted NSR path:%s", path)
3606
3607
3608 class VnfrDtsHandler(object):
3609     """ The virtual network service DTS handler """
3610     XPATH = "D,/vnfr:vnfr-catalog/vnfr:vnfr"
3611
3612     def __init__(self, dts, log, loop, nsm):
3613         self._dts = dts
3614         self._log = log
3615         self._loop = loop
3616         self._nsm = nsm
3617
3618         self._regh = None
3619
3620     @property
3621     def regh(self):
3622         """ Return registration handle """
3623         return self._regh
3624
3625     @property
3626     def nsm(self):
3627         """ Return the NS manager instance """
3628         return self._nsm
3629
3630     @asyncio.coroutine
3631     def register(self):
3632         """ Register for vnfr create/update/delete/ advises from dts """
3633         if self._regh:
3634             self._log.warning("VNFR DTS handler already registered for project {}".
3635                               format(self._project.name))
3636             return
3637
3638
3639         def on_commit(xact_info):
3640             """ The transaction has been committed """
3641             self._log.debug("Got vnfr commit (xact_info: %s)", xact_info)
3642             return rwdts.MemberRspCode.ACTION_OK
3643
3644         @asyncio.coroutine
3645         def on_prepare(xact_info, action, ks_path, msg):
3646             """ prepare callback from dts """
3647             xpath = ks_path.to_xpath(RwNsrYang.get_schema())
3648             self._log.debug(
3649                 "Got vnfr on_prepare cb (xact_info: %s, action: %s): %s:%s",
3650                 xact_info, action, ks_path, msg
3651                 )
3652
3653             schema = VnfrYang.YangData_RwProject_Project_VnfrCatalog_Vnfr.schema()
3654             path_entry = schema.keyspec_to_entry(ks_path)
3655             if path_entry.key00.id not in self._nsm._vnfrs:
3656                 # Check if this is a monitoring param xpath
3657                 if 'vnfr:monitoring-param' not in xpath:
3658                     self._log.error("%s request for non existent record path %s",
3659                                     action, xpath)
3660                 xact_info.respond_xpath(rwdts.XactRspCode.NA, xpath)
3661
3662                 return
3663
3664             if action == rwdts.QueryAction.CREATE or action == rwdts.QueryAction.UPDATE:
3665                 yield from self._nsm.update_vnfr(msg)
3666             elif action == rwdts.QueryAction.DELETE:
3667                 self._log.debug("Deleting VNFR with id %s", path_entry.key00.id)
3668                 self._nsm.delete_vnfr(path_entry.key00.id)
3669
3670             xact_info.respond_xpath(rwdts.XactRspCode.ACK, xpath)
3671
3672         self._log.debug("Registering for VNFR using xpath: %s",
3673                         VnfrDtsHandler.XPATH,)
3674
3675         hdl = rift.tasklets.DTS.RegistrationHandler(on_commit=on_commit,
3676                                                     on_prepare=on_prepare,)
3677         with self._dts.group_create() as group:
3678             self._regh = group.register(xpath=self._nsm._project.add_project(
3679                 VnfrDtsHandler.XPATH),
3680                                         handler=hdl,
3681                                         flags=(rwdts.Flag.SUBSCRIBER),)
3682
3683     def deregister(self):
3684         self._log.debug("De-register VNFR for project {}".
3685                         format(self._nsm._project.name))
3686         if self._regh:
3687             self._regh.deregister()
3688             self._regh = None
3689
3690 class NsdRefCountDtsHandler(object):
3691     """ The NSD Ref Count DTS handler """
3692     XPATH = "D,/nsr:ns-instance-opdata/rw-nsr:nsd-ref-count"
3693
3694     def __init__(self, dts, log, loop, nsm):
3695         self._dts = dts
3696         self._log = log
3697         self._loop = loop
3698         self._nsm = nsm
3699
3700         self._regh = None
3701
3702     @property
3703     def regh(self):
3704         """ Return registration handle """
3705         return self._regh
3706
3707     @property
3708     def nsm(self):
3709         """ Return the NS manager instance """
3710         return self._nsm
3711
3712     @asyncio.coroutine
3713     def register(self):
3714         """ Register for NSD ref count read from dts """
3715         if self._regh:
3716             self._log.warning("NSD ref DTS handler already registered for project {}".
3717                               format(self._project.name))
3718             return
3719
3720
3721         @asyncio.coroutine
3722         def on_prepare(xact_info, action, ks_path, msg):
3723             """ prepare callback from dts """
3724             xpath = ks_path.to_xpath(RwNsrYang.get_schema())
3725
3726             if action == rwdts.QueryAction.READ:
3727                 schema = RwNsrYang.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount.schema()
3728                 path_entry = schema.keyspec_to_entry(ks_path)
3729                 nsd_list = yield from self._nsm.get_nsd_refcount(path_entry.key00.nsd_id_ref)
3730                 for xpath, msg in nsd_list:
3731                     xact_info.respond_xpath(rsp_code=rwdts.XactRspCode.MORE,
3732                                             xpath=xpath,
3733                                             msg=msg)
3734                 xact_info.respond_xpath(rwdts.XactRspCode.ACK)
3735             else:
3736                 raise NetworkServiceRecordError("Not supported operation %s" % action)
3737
3738         hdl = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_prepare,)
3739         with self._dts.group_create() as group:
3740             self._regh = group.register(xpath=self._nsm._project.add_project(
3741                 NsdRefCountDtsHandler.XPATH),
3742                                         handler=hdl,
3743                                         flags=rwdts.Flag.PUBLISHER,)
3744
3745     def deregister(self):
3746         self._log.debug("De-register NSD Ref count for project {}".
3747                         format(self._nsm._project.name))
3748         if self._regh:
3749             self._regh.deregister()
3750             self._regh = None
3751
3752
3753 class NsManager(object):
3754     """ The Network Service Manager class"""
3755     def __init__(self, dts, log, loop, project,
3756                  nsr_handler, vnfr_handler, vlr_handler, ro_plugin_selector,
3757                  vnffgmgr, vnfd_pub_handler, cloud_account_handler):
3758         self._dts = dts
3759         self._log = log
3760         self._loop = loop
3761         self._project = project
3762         self._nsr_handler = nsr_handler
3763         self._vnfr_pub_handler = vnfr_handler
3764         self._vlr_pub_handler = vlr_handler
3765         self._vnffgmgr = vnffgmgr
3766         self._vnfd_pub_handler = vnfd_pub_handler
3767         self._cloud_account_handler = cloud_account_handler
3768
3769         self._ro_plugin_selector = ro_plugin_selector
3770         self._ncclient = rift.mano.ncclient.NcClient(
3771               host="127.0.0.1",
3772               port=2022,
3773               username="admin",
3774               password="admin",
3775               loop=self._loop)
3776
3777         self._nsrs = {}
3778         self._nsds = {}
3779         self._vnfds = {}
3780         self._vnfrs = {}
3781
3782         self.cfgmgr_obj = conman.ROConfigManager(log, loop, dts, self)
3783
3784         # TODO: All these handlers should move to tasklet level.
3785         # Passing self is often an indication of bad design
3786         self._nsd_dts_handler = NsdDtsHandler(dts, log, loop, self)
3787         self._vnfd_dts_handler = VnfdDtsHandler(dts, log, loop, self)
3788         self._dts_handlers = [self._nsd_dts_handler,
3789                               VnfrDtsHandler(dts, log, loop, self),
3790                               NsdRefCountDtsHandler(dts, log, loop, self),
3791                               NsrDtsHandler(dts, log, loop, self),
3792                               ScalingRpcHandler(log, dts, loop, self._project,
3793                                                 self.scale_rpc_callback),
3794                               NsrRpcDtsHandler(dts, log, loop, self),
3795                               self._vnfd_dts_handler,
3796                               self.cfgmgr_obj,
3797                               ]
3798
3799
3800     @property
3801     def log(self):
3802         """ Log handle """
3803         return self._log
3804
3805     @property
3806     def loop(self):
3807         """ Loop """
3808         return self._loop
3809
3810     @property
3811     def dts(self):
3812         """ DTS handle """
3813         return self._dts
3814
3815     @property
3816     def nsr_handler(self):
3817         """" NSR handler """
3818         return self._nsr_handler
3819
3820     @property
3821     def so_obj(self):
3822         """" So Obj handler """
3823         return self._so_obj
3824
3825     @property
3826     def nsrs(self):
3827         """ NSRs in this NSM"""
3828         return self._nsrs
3829
3830     @property
3831     def nsds(self):
3832         """ NSDs in this NSM"""
3833         return self._nsds
3834
3835     @property
3836     def vnfds(self):
3837         """ VNFDs in this NSM"""
3838         return self._vnfds
3839
3840     @property
3841     def vnfrs(self):
3842         """ VNFRs in this NSM"""
3843         return self._vnfrs
3844
3845     @property
3846     def nsr_pub_handler(self):
3847         """ NSR publication handler """
3848         return self._nsr_handler
3849
3850     @property
3851     def vnfr_pub_handler(self):
3852         """ VNFR publication handler """
3853         return self._vnfr_pub_handler
3854
3855     @property
3856     def vlr_pub_handler(self):
3857         """ VLR publication handler """
3858         return self._vlr_pub_handler
3859
3860     @property
3861     def vnfd_pub_handler(self):
3862         return self._vnfd_pub_handler
3863
3864     @asyncio.coroutine
3865     def register(self):
3866         """ Register all static DTS handlers """
3867         for dts_handle in self._dts_handlers:
3868             yield from dts_handle.register()
3869
3870     def deregister(self):
3871         """ Register all static DTS handlers """
3872         for dts_handle in self._dts_handlers:
3873             dts_handle.deregister()
3874
3875
3876     def get_ns_by_nsr_id(self, nsr_id):
3877         """ get NSR by nsr id """
3878         if nsr_id not in self._nsrs:
3879             raise NetworkServiceRecordError("NSR id %s not found" % nsr_id)
3880
3881         return self._nsrs[nsr_id]
3882
3883     def scale_nsr_out(self, nsr_id, scale_group_name, instance_id, config_xact):
3884         self.log.debug("Scale out NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3885                        nsr_id,
3886                        scale_group_name,
3887                        instance_id
3888                        )
3889         nsr = self._nsrs[nsr_id]
3890         if nsr.state != NetworkServiceRecordState.RUNNING:
3891             raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3892
3893         self._loop.create_task(nsr.create_scale_group_instance(scale_group_name, instance_id, config_xact))
3894
3895     def scale_nsr_in(self, nsr_id, scale_group_name, instance_id):
3896         self.log.debug("Scale in NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3897                        nsr_id,
3898                        scale_group_name,
3899                        instance_id,
3900                        )
3901         nsr = self._nsrs[nsr_id]
3902         if nsr.state != NetworkServiceRecordState.RUNNING:
3903             raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3904
3905         self._loop.create_task(nsr.delete_scale_group_instance(scale_group_name, instance_id))
3906
3907     def scale_rpc_callback(self, xact, msg, action):
3908         """Callback handler for RPC calls
3909         Args:
3910             xact : Transaction Handler
3911             msg : RPC input
3912             action : Scaling Action
3913         """
3914         ScalingGroupInstance = NsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup_Instance
3915         ScalingGroup = NsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup
3916
3917         xpath = self._project.add_project(
3918             ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').
3919             format(msg.nsr_id_ref))
3920
3921         instance = ScalingGroupInstance.from_dict({
3922             "id": msg.instance_id,
3923             "project_name": self._project.name,})
3924
3925         @asyncio.coroutine
3926         def get_nsr_scaling_group():
3927             results = yield from self._dts.query_read(xpath, rwdts.XactFlag.MERGE)
3928
3929             for result in results:
3930                 res = yield from result
3931                 nsr_config = res.result
3932
3933             for scaling_group in nsr_config.scaling_group:
3934                 if scaling_group.scaling_group_name_ref == msg.scaling_group_name_ref:
3935                     break
3936             else:
3937                 scaling_group = nsr_config.scaling_group.add()
3938                 scaling_group.scaling_group_name_ref = msg.scaling_group_name_ref
3939
3940             return (nsr_config, scaling_group)
3941
3942         @asyncio.coroutine
3943         def update_config(nsr_config):
3944             xml = self._ncclient.convert_to_xml(RwNsrYang, nsr_config)
3945             xml = '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml)
3946             yield from self._ncclient.connect()
3947             yield from self._ncclient.manager.edit_config(target="running", config=xml, default_operation="replace")
3948
3949         @asyncio.coroutine
3950         def scale_out():
3951             nsr_config, scaling_group = yield from get_nsr_scaling_group()
3952             scaling_group.instance.append(instance)
3953             yield from update_config(nsr_config)
3954
3955         @asyncio.coroutine
3956         def scale_in():
3957             nsr_config, scaling_group = yield from get_nsr_scaling_group()
3958             scaling_group.instance.remove(instance)
3959             yield from update_config(nsr_config)
3960
3961         if action == ScalingRpcHandler.ACTION.SCALE_OUT:
3962             self._loop.create_task(scale_out())
3963         else:
3964             self._loop.create_task(scale_in())
3965
3966         # Opdata based calls, disabled for now!
3967         # if action == ScalingRpcHandler.ACTION.SCALE_OUT:
3968         #     self.scale_nsr_out(
3969         #           msg.nsr_id_ref,
3970         #           msg.scaling_group_name_ref,
3971         #           msg.instance_id,
3972         #           xact)
3973         # else:
3974         #     self.scale_nsr_in(
3975         #           msg.nsr_id_ref,
3976         #           msg.scaling_group_name_ref,
3977         #           msg.instance_id)
3978
3979     def nsr_update_cfg(self, nsr_id, msg):
3980         nsr = self._nsrs[nsr_id]
3981         nsr.nsr_cfg_msg= msg
3982
3983     def nsr_instantiate_vl(self, nsr_id, vld):
3984         self.log.debug("NSR {} create VL {}".format(nsr_id, vld))
3985         nsr = self._nsrs[nsr_id]
3986         if nsr.state != NetworkServiceRecordState.RUNNING:
3987             raise NsrVlUpdateError("Cannot perform VL instantiate if NSR is not in running state")
3988
3989         # Not calling in a separate task as this is called from a separate task
3990         yield from nsr.create_vl_instance(vld)
3991
3992     def nsr_terminate_vl(self, nsr_id, vld):
3993         self.log.debug("NSR {} delete VL {}".format(nsr_id, vld.id))
3994         nsr = self._nsrs[nsr_id]
3995         if nsr.state != NetworkServiceRecordState.RUNNING:
3996             raise NsrVlUpdateError("Cannot perform VL terminate if NSR is not in running state")
3997
3998         # Not calling in a separate task as this is called from a separate task
3999         yield from nsr.delete_vl_instance(vld)
4000
4001     def create_nsr(self, nsr_msg, key_pairs=None,restart_mode=False):
4002         """ Create an NSR instance """
4003         if nsr_msg.id in self._nsrs:
4004             msg = "NSR id %s already exists" % nsr_msg.id
4005             self._log.error(msg)
4006             raise NetworkServiceRecordError(msg)
4007
4008         self._log.info("Create NetworkServiceRecord nsr id %s from nsd_id %s",
4009                        nsr_msg.id,
4010                        nsr_msg.nsd.id)
4011
4012         nsm_plugin = self._ro_plugin_selector.ro_plugin
4013         sdn_account_name = self._cloud_account_handler.get_cloud_account_sdn_name(nsr_msg.cloud_account)
4014
4015         nsr = NetworkServiceRecord(self._dts,
4016                                    self._log,
4017                                    self._loop,
4018                                    self,
4019                                    nsm_plugin,
4020                                    nsr_msg,
4021                                    sdn_account_name,
4022                                    key_pairs,
4023                                    self._project,
4024                                    restart_mode=restart_mode,
4025                                    vlr_handler=self._ro_plugin_selector._records_publisher._vlr_pub_hdlr
4026                                    )
4027         self._nsrs[nsr_msg.id] = nsr
4028         nsm_plugin.create_nsr(nsr_msg, nsr_msg.nsd, key_pairs)
4029
4030         return nsr
4031
4032     def delete_nsr(self, nsr_id):
4033         """
4034         Delete NSR with the passed nsr id
4035         """
4036         del self._nsrs[nsr_id]
4037
4038     @asyncio.coroutine
4039     def instantiate_ns(self, nsr_id, config_xact):
4040         """ Instantiate an NS instance """
4041         self._log.debug("Instantiating Network service id %s", nsr_id)
4042         if nsr_id not in self._nsrs:
4043             err = "NSR id %s not found " % nsr_id
4044             self._log.error(err)
4045             raise NetworkServiceRecordError(err)
4046
4047         nsr = self._nsrs[nsr_id]
4048         yield from nsr.nsm_plugin.instantiate_ns(nsr, config_xact)
4049
4050     @asyncio.coroutine
4051     def update_vnfr(self, vnfr):
4052         """Create/Update an VNFR """
4053
4054         vnfr_state = self._vnfrs[vnfr.id].state
4055         self._log.debug("Updating VNFR with state %s: vnfr %s", vnfr_state, vnfr)
4056
4057         yield from self._vnfrs[vnfr.id].update_state(vnfr)
4058         nsr = self.find_nsr_for_vnfr(vnfr.id)
4059         yield from nsr.update_state()
4060
4061     def find_nsr_for_vnfr(self, vnfr_id):
4062         """ Find the NSR which )has the passed vnfr id"""
4063         for nsr in list(self.nsrs.values()):
4064             for vnfr in list(nsr.vnfrs.values()):
4065                 if vnfr.id == vnfr_id:
4066                     return nsr
4067         return None
4068
4069     def delete_vnfr(self, vnfr_id):
4070         """ Delete VNFR  with the passed id"""
4071         del self._vnfrs[vnfr_id]
4072
4073     def get_nsd_ref(self, nsd_id):
4074         """ Get network service descriptor for the passed nsd_id
4075             with a reference"""
4076         nsd = self.get_nsd(nsd_id)
4077         nsd.ref()
4078         return nsd
4079
4080     @asyncio.coroutine
4081     def get_nsr_config(self, nsd_id):
4082         xpath = self._project.add_project("C,/nsr:ns-instance-config")
4083         results = yield from self._dts.query_read(xpath, rwdts.XactFlag.MERGE)
4084
4085         for result in results:
4086             entry = yield from result
4087             ns_instance_config = entry.result
4088
4089             for nsr in ns_instance_config.nsr:
4090                 if nsr.nsd.id == nsd_id:
4091                     return nsr
4092
4093         return None
4094
4095     @asyncio.coroutine
4096     def nsd_unref_by_nsr_id(self, nsr_id):
4097         """ Unref the network service descriptor based on NSR id """
4098         self._log.debug("NSR Unref called for Nsr Id:%s", nsr_id)
4099         if nsr_id in self._nsrs:
4100             nsr = self._nsrs[nsr_id]
4101
4102             try:
4103                 nsd = self.get_nsd(nsr.nsd_id)
4104                 self._log.debug("Releasing ref on NSD %s held by NSR %s - Curr %d",
4105                                 nsd.id, nsr.id, nsd.ref_count)
4106                 nsd.unref()
4107             except NetworkServiceDescriptorError:
4108                 # We store a copy of NSD in NSR and the NSD in nsd-catalog
4109                 # could be deleted
4110                 pass
4111
4112         else:
4113             self._log.error("Cannot find NSR with id %s", nsr_id)
4114             raise NetworkServiceDescriptorUnrefError("No NSR with id" % nsr_id)
4115
4116     @asyncio.coroutine
4117     def nsd_unref(self, nsd_id):
4118         """ Unref the network service descriptor associated with the id """
4119         nsd = self.get_nsd(nsd_id)
4120         nsd.unref()
4121
4122     def get_nsd(self, nsd_id):
4123         """ Get network service descriptor for the passed nsd_id"""
4124         if nsd_id not in self._nsds:
4125             self._log.error("Cannot find NSD id:%s", nsd_id)
4126             raise NetworkServiceDescriptorError("Cannot find NSD id:%s", nsd_id)
4127
4128         return self._nsds[nsd_id]
4129
4130     def create_nsd(self, nsd_msg):
4131         """ Create a network service descriptor """
4132         self._log.debug("Create network service descriptor - %s", nsd_msg)
4133         if nsd_msg.id in self._nsds:
4134             self._log.error("Cannot create NSD %s -NSD ID already exists", nsd_msg)
4135             raise NetworkServiceDescriptorError("NSD already exists-%s", nsd_msg.id)
4136
4137         nsd = NetworkServiceDescriptor(
4138                 self._dts,
4139                 self._log,
4140                 self._loop,
4141                 nsd_msg,
4142                 self
4143                 )
4144         self._nsds[nsd_msg.id] = nsd
4145
4146         return nsd
4147
4148     def update_nsd(self, nsd):
4149         """ update the Network service descriptor """
4150         self._log.debug("Update network service descriptor - %s", nsd)
4151         if nsd.id not in self._nsds:
4152             self._log.debug("No NSD found - creating NSD id = %s", nsd.id)
4153             self.create_nsd(nsd)
4154         else:
4155             self._log.debug("Updating NSD id = %s, nsd = %s", nsd.id, nsd)
4156             self._nsds[nsd.id].update(nsd)
4157
4158     def delete_nsd(self, nsd_id):
4159         """ Delete the Network service descriptor with the passed id """
4160         self._log.debug("Deleting the network service descriptor - %s", nsd_id)
4161         if nsd_id not in self._nsds:
4162             self._log.debug("Delete NSD failed - cannot find nsd-id %s", nsd_id)
4163             raise NetworkServiceDescriptorNotFound("Cannot find %s", nsd_id)
4164
4165         if nsd_id not in self._nsds:
4166             self._log.debug("Cannot delete NSD id %s reference exists %s",
4167                             nsd_id,
4168                             self._nsds[nsd_id].ref_count)
4169             raise NetworkServiceDescriptorRefCountExists(
4170                 "Cannot delete :%s, ref_count:%s",
4171                 nsd_id,
4172                 self._nsds[nsd_id].ref_count)
4173
4174         del self._nsds[nsd_id]
4175
4176     def get_vnfd_config(self, xact):
4177         vnfd_dts_reg = self._vnfd_dts_handler.regh
4178         for cfg in vnfd_dts_reg.get_xact_elements(xact):
4179             if cfg.id not in self._vnfds:
4180                 self.create_vnfd(cfg)
4181
4182     def get_vnfd(self, vnfd_id, xact):
4183         """ Get virtual network function descriptor for the passed vnfd_id"""
4184         if vnfd_id not in self._vnfds:
4185             self._log.error("Cannot find VNFD id:%s", vnfd_id)
4186             self.get_vnfd_config(xact)
4187
4188             if vnfd_id not in self._vnfds:
4189                 self._log.error("Cannot find VNFD id:%s", vnfd_id)
4190                 raise VnfDescriptorError("Cannot find VNFD id:%s", vnfd_id)
4191
4192         return self._vnfds[vnfd_id]
4193
4194     def create_vnfd(self, vnfd):
4195         """ Create a virtual network function descriptor """
4196         self._log.debug("Create virtual network function descriptor - %s", vnfd)
4197         if vnfd.id in self._vnfds:
4198             self._log.error("Cannot create VNFD %s -VNFD ID already exists", vnfd)
4199             raise VnfDescriptorError("VNFD already exists-%s", vnfd.id)
4200
4201         self._vnfds[vnfd.id] = vnfd
4202         return self._vnfds[vnfd.id]
4203
4204     def update_vnfd(self, vnfd):
4205         """ Update the virtual network function descriptor """
4206         self._log.debug("Update virtual network function descriptor- %s", vnfd)
4207
4208
4209         if vnfd.id not in self._vnfds:
4210             self._log.debug("No VNFD found - creating VNFD id = %s", vnfd.id)
4211             self.create_vnfd(vnfd)
4212         else:
4213             self._log.debug("Updating VNFD id = %s, vnfd = %s", vnfd.id, vnfd)
4214             self._vnfds[vnfd.id] = vnfd
4215
4216     @asyncio.coroutine
4217     def delete_vnfd(self, vnfd_id):
4218         """ Delete the virtual network function descriptor with the passed id """
4219         self._log.debug("Deleting the virtual network function descriptor - %s", vnfd_id)
4220         if vnfd_id not in self._vnfds:
4221             self._log.debug("Delete VNFD failed - cannot find vnfd-id %s", vnfd_id)
4222             raise VnfDescriptorError("Cannot find %s", vnfd_id)
4223
4224         del self._vnfds[vnfd_id]
4225
4226     def nsd_in_use(self, nsd_id):
4227         """ Is the NSD with the passed id in use """
4228         self._log.debug("Is this NSD in use - msg:%s", nsd_id)
4229         if nsd_id in self._nsds:
4230             return self._nsds[nsd_id].in_use()
4231         return False
4232
4233     @asyncio.coroutine
4234     def publish_nsr(self, xact, path, msg):
4235         """ Publish a NSR """
4236         self._log.debug("Publish NSR with path %s, msg %s",
4237                         path, msg)
4238         yield from self.nsr_handler.update(xact, path, msg)
4239
4240     @asyncio.coroutine
4241     def unpublish_nsr(self, xact, path):
4242         """ Un Publish an NSR """
4243         self._log.debug("Publishing delete NSR with path %s", path)
4244         yield from self.nsr_handler.delete(path, xact)
4245
4246     def vnfr_is_ready(self, vnfr_id):
4247         """ VNFR with the id is ready """
4248         self._log.debug("VNFR id %s ready", vnfr_id)
4249         if vnfr_id not in self._vnfds:
4250             err = "Did not find VNFR ID with id %s" % vnfr_id
4251             self._log.critical("err")
4252             raise VirtualNetworkFunctionRecordError(err)
4253         self._vnfrs[vnfr_id].is_ready()
4254
4255     @asyncio.coroutine
4256     def get_nsd_refcount(self, nsd_id):
4257         """ Get the nsd_list from this NSM"""
4258
4259         def nsd_refcount_xpath(nsd_id):
4260             """ xpath for ref count entry """
4261             return (self._project.add_project(NsdRefCountDtsHandler.XPATH) +
4262                     "[rw-nsr:nsd-id-ref = '{}']").format(nsd_id)
4263
4264         nsd_list = []
4265         if nsd_id is None or nsd_id == "":
4266             for nsd in self._nsds.values():
4267                 nsd_msg = RwNsrYang.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount()
4268                 nsd_msg.nsd_id_ref = nsd.id
4269                 nsd_msg.instance_ref_count = nsd.ref_count
4270                 nsd_list.append((nsd_refcount_xpath(nsd.id), nsd_msg))
4271         elif nsd_id in self._nsds:
4272             nsd_msg = RwNsrYang.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount()
4273             nsd_msg.nsd_id_ref = self._nsds[nsd_id].id
4274             nsd_msg.instance_ref_count = self._nsds[nsd_id].ref_count
4275             nsd_list.append((nsd_refcount_xpath(nsd_id), nsd_msg))
4276
4277         return nsd_list
4278
4279     @asyncio.coroutine
4280     def terminate_ns(self, nsr_id, xact):
4281         """
4282         Terminate network service for the given NSR Id
4283         """
4284
4285         # Terminate the instances/networks assocaited with this nw service
4286         self._log.debug("Terminating the network service %s", nsr_id)
4287         try :
4288             yield from self._nsrs[nsr_id].terminate()
4289         except Exception as e:
4290             self.log.exception("Failed to terminate NSR[id=%s]", nsr_id)
4291
4292         # Unref the NSD
4293         yield from self.nsd_unref_by_nsr_id(nsr_id)
4294
4295         # Unpublish the NSR record
4296         self._log.debug("Unpublishing the network service %s", nsr_id)
4297         yield from self._nsrs[nsr_id].unpublish(xact)
4298
4299         # Finaly delete the NS instance from this NS Manager
4300         self._log.debug("Deletng the network service %s", nsr_id)
4301         self.delete_nsr(nsr_id)
4302
4303
4304 class NsmRecordsPublisherProxy(object):
4305     """ This class provides a publisher interface that allows plugin objects
4306         to publish NSR/VNFR/VLR"""
4307
4308     def __init__(self, dts, log, loop, project, nsr_pub_hdlr,
4309                  vnfr_pub_hdlr, vlr_pub_hdlr,):
4310         self._dts = dts
4311         self._log = log
4312         self._loop = loop
4313         self._project = project
4314         self._nsr_pub_hdlr = nsr_pub_hdlr
4315         self._vlr_pub_hdlr = vlr_pub_hdlr
4316         self._vnfr_pub_hdlr = vnfr_pub_hdlr
4317
4318     @asyncio.coroutine
4319     def publish_nsr(self, xact, nsr):
4320         """ Publish an NSR """
4321         path = NetworkServiceRecord.xpath_from_nsr(nsr)
4322         return (yield from self._nsr_pub_hdlr.update(xact, path, nsr))
4323
4324     @asyncio.coroutine
4325     def unpublish_nsr(self, xact, nsr):
4326         """ Unpublish an NSR """
4327         path = NetworkServiceRecord.xpath_from_nsr(nsr)
4328         return (yield from self._nsr_pub_hdlr.delete(xact, path))
4329
4330     @asyncio.coroutine
4331     def publish_vnfr(self, xact, vnfr):
4332         """ Publish an VNFR """
4333         path = VirtualNetworkFunctionRecord.vnfr_xpath(vnfr)
4334         return (yield from self._vnfr_pub_hdlr.update(xact, path, vnfr))
4335
4336     @asyncio.coroutine
4337     def unpublish_vnfr(self, xact, vnfr):
4338         """ Unpublish a VNFR """
4339         path = VirtualNetworkFunctionRecord.vnfr_xpath(vnfr)
4340         return (yield from self._vnfr_pub_hdlr.delete(xact, path))
4341
4342     @asyncio.coroutine
4343     def publish_vlr(self, xact, vlr):
4344         """ Publish a VLR """
4345         path = self._project.add_project(VirtualLinkRecord.vlr_xpath(vlr))
4346         return (yield from self._vlr_pub_hdlr.update(xact, path, vlr))
4347
4348     @asyncio.coroutine
4349     def unpublish_vlr(self, xact, vlr):
4350         """ Unpublish a VLR """
4351         path = self._project.add_project(VirtualLinkRecord.vlr_xpath(vlr))
4352         return (yield from self._vlr_pub_hdlr.delete(xact, path))
4353
4354
4355 class ScalingRpcHandler(mano_dts.DtsHandler):
4356     """ The Network service Monitor DTS handler """
4357     SCALE_IN_INPUT_XPATH = "I,/nsr:exec-scale-in"
4358     SCALE_IN_OUTPUT_XPATH = "O,/nsr:exec-scale-in"
4359
4360     SCALE_OUT_INPUT_XPATH = "I,/nsr:exec-scale-out"
4361     SCALE_OUT_OUTPUT_XPATH = "O,/nsr:exec-scale-out"
4362
4363     ACTION = Enum('ACTION', 'SCALE_IN SCALE_OUT')
4364
4365     def __init__(self, log, dts, loop, project, callback=None):
4366         super().__init__(log, dts, loop, project)
4367         self.callback = callback
4368         self.last_instance_id = defaultdict(int)
4369         self._regh_in = None
4370         self._regh_out = None
4371
4372     @asyncio.coroutine
4373     def register(self):
4374
4375         if self._regh_in:
4376             self._log.warning("RPC already registered for project {}".
4377                               format(self._project.name))
4378             return
4379
4380         @asyncio.coroutine
4381         def on_scale_in_prepare(xact_info, action, ks_path, msg):
4382             assert action == rwdts.QueryAction.RPC
4383
4384             try:
4385                 if not self._project.rpc_check(msg, xact_info=xact_info):
4386                     return
4387
4388                 if self.callback:
4389                     self.callback(xact_info.xact, msg, self.ACTION.SCALE_IN)
4390
4391                 rpc_op = NsrYang.YangOutput_Nsr_ExecScaleIn.from_dict({
4392                     "instance_id": msg.instance_id,
4393                     "project_name": self._project.name,})
4394
4395                 xact_info.respond_xpath(
4396                     rwdts.XactRspCode.ACK,
4397                     self.__class__.SCALE_IN_OUTPUT_XPATH,
4398                     rpc_op)
4399
4400             except Exception as e:
4401                 self.log.exception(e)
4402                 xact_info.respond_xpath(
4403                     rwdts.XactRspCode.NACK,
4404                     self.__class__.SCALE_IN_OUTPUT_XPATH)
4405
4406         @asyncio.coroutine
4407         def on_scale_out_prepare(xact_info, action, ks_path, msg):
4408             assert action == rwdts.QueryAction.RPC
4409
4410             try:
4411                 if not self._project.rpc_check(msg, xact_info=xact_info):
4412                     return
4413
4414                 scaling_group = msg.scaling_group_name_ref
4415                 if not msg.instance_id:
4416                     last_instance_id = self.last_instance_id[scale_group]
4417                     msg.instance_id  = last_instance_id + 1
4418                     self.last_instance_id[scale_group] += 1
4419
4420                 if self.callback:
4421                     self.callback(xact_info.xact, msg, self.ACTION.SCALE_OUT)
4422
4423                 rpc_op = NsrYang.YangOutput_Nsr_ExecScaleOut.from_dict({
4424                     "instance_id": msg.instance_id,
4425                     "project_name": self._project.name,})
4426
4427                 xact_info.respond_xpath(
4428                     rwdts.XactRspCode.ACK,
4429                     self.__class__.SCALE_OUT_OUTPUT_XPATH,
4430                     rpc_op)
4431
4432             except Exception as e:
4433                 self.log.exception(e)
4434                 xact_info.respond_xpath(
4435                       rwdts.XactRspCode.NACK,
4436                       self.__class__.SCALE_OUT_OUTPUT_XPATH)
4437
4438         scale_in_hdl = rift.tasklets.DTS.RegistrationHandler(
4439               on_prepare=on_scale_in_prepare)
4440         scale_out_hdl = rift.tasklets.DTS.RegistrationHandler(
4441               on_prepare=on_scale_out_prepare)
4442
4443         with self.dts.group_create() as group:
4444             self._regh_in = group.register(
4445                 xpath=self.__class__.SCALE_IN_INPUT_XPATH,
4446                 handler=scale_in_hdl,
4447                 flags=rwdts.Flag.PUBLISHER)
4448             self._regh_out = group.register(
4449                 xpath=self.__class__.SCALE_OUT_INPUT_XPATH,
4450                 handler=scale_out_hdl,
4451                 flags=rwdts.Flag.PUBLISHER)
4452
4453     def deregister(self):
4454         self._log.debug("De-register scale RPCs for project {}".
4455                         format(self._project.name))
4456         if self._regh_in:
4457             self._regh_in.deregister()
4458             self._regh_in = None
4459         if self._regh_out:
4460             self._regh_out.deregister()
4461             self._regh_out = None
4462
4463
4464 class NsmProject(ManoProject):
4465
4466     def __init__(self, name, tasklet, **kw):
4467         super(NsmProject, self).__init__(tasklet.log, name)
4468         self.update(tasklet)
4469
4470         self._nsm = None
4471
4472         self._ro_plugin_selector = None
4473         self._vnffgmgr = None
4474
4475         self._nsr_pub_handler = None
4476         self._vnfr_pub_handler = None
4477         self._vlr_pub_handler = None
4478         self._vnfd_pub_handler = None
4479         self._scale_cfg_handler = None
4480
4481         self._records_publisher_proxy = None
4482
4483     @asyncio.coroutine
4484     def register(self):
4485         self._nsr_pub_handler = publisher.NsrOpDataDtsHandler(
4486             self._dts, self.log, self.loop, self)
4487         yield from self._nsr_pub_handler.register()
4488
4489         self._vnfr_pub_handler = publisher.VnfrPublisherDtsHandler(
4490             self._dts, self.log, self.loop, self)
4491         yield from self._vnfr_pub_handler.register()
4492
4493         self._vlr_pub_handler = publisher.VlrPublisherDtsHandler(
4494             self._dts, self.log, self.loop, self)
4495         yield from self._vlr_pub_handler.register()
4496
4497         manifest = self._tasklet.tasklet_info.get_pb_manifest()
4498         use_ssl = manifest.bootstrap_phase.rwsecurity.use_ssl
4499         ssl_cert = manifest.bootstrap_phase.rwsecurity.cert
4500         ssl_key = manifest.bootstrap_phase.rwsecurity.key
4501
4502         self._vnfd_pub_handler = publisher.VnfdPublisher(
4503             use_ssl, ssl_cert, ssl_key, self.loop, self)
4504
4505         self._records_publisher_proxy = NsmRecordsPublisherProxy(
4506                 self._dts,
4507                 self.log,
4508                 self.loop,
4509                 self,
4510                 self._nsr_pub_handler,
4511                 self._vnfr_pub_handler,
4512                 self._vlr_pub_handler,
4513                 )
4514
4515         # Register the NSM to receive the nsm plugin
4516         # when cloud account is configured
4517         self._ro_plugin_selector = cloud.ROAccountPluginSelector(
4518                 self._dts,
4519                 self.log,
4520                 self.loop,
4521                 self,
4522                 self._records_publisher_proxy,
4523                 )
4524         yield from self._ro_plugin_selector.register()
4525
4526         self._cloud_account_handler = cloud.CloudAccountConfigSubscriber(
4527                 self._log,
4528                 self._dts,
4529                 self.log_hdl,
4530                 self,
4531                 )
4532
4533         yield from self._cloud_account_handler.register()
4534
4535         self._vnffgmgr = rwvnffgmgr.VnffgMgr(self._dts, self.log, self.log_hdl, self.loop, self)
4536         yield from self._vnffgmgr.register()
4537
4538         self._nsm = NsManager(
4539                 self._dts,
4540                 self.log,
4541                 self.loop,
4542                 self,
4543                 self._nsr_pub_handler,
4544                 self._vnfr_pub_handler,
4545                 self._vlr_pub_handler,
4546                 self._ro_plugin_selector,
4547                 self._vnffgmgr,
4548                 self._vnfd_pub_handler,
4549                 self._cloud_account_handler,
4550                 )
4551
4552         yield from self._nsm.register()
4553
4554     def deregister(self):
4555         self._log.debug("Project {} de-register".format(self.name))
4556         self._nsm.deregister()
4557         self._vnffgmgr.deregister()
4558         self._cloud_account_handler.deregister()
4559         self._ro_plugin_selector.deregister()
4560         self._nsm = None
4561
4562     @asyncio.coroutine
4563     def delete_prepare(self):
4564         # Check if any NS instance is present
4565         if self._nsm and self._nsm._nsrs:
4566             return False
4567         return True
4568
4569
4570 class NsmTasklet(rift.tasklets.Tasklet):
4571     """
4572     The network service manager  tasklet
4573     """
4574     def __init__(self, *args, **kwargs):
4575         super(NsmTasklet, self).__init__(*args, **kwargs)
4576         self.rwlog.set_category("rw-mano-log")
4577         self.rwlog.set_subcategory("nsm")
4578
4579         self._dts = None
4580         self.project_handler = None
4581         self.projects = {}
4582
4583     @property
4584     def dts(self):
4585         return self._dts
4586
4587     def start(self):
4588         """ The task start callback """
4589         super(NsmTasklet, self).start()
4590         self.log.info("Starting NsmTasklet")
4591
4592         self.log.debug("Registering with dts")
4593         self._dts = rift.tasklets.DTS(self.tasklet_info,
4594                                       RwNsmYang.get_schema(),
4595                                       self.loop,
4596                                       self.on_dts_state_change)
4597
4598         self.log.debug("Created DTS Api GI Object: %s", self._dts)
4599
4600     def stop(self):
4601         try:
4602             self._dts.deinit()
4603         except Exception:
4604             print("Caught Exception in NSM stop:", sys.exc_info()[0])
4605             raise
4606
4607     def on_instance_started(self):
4608         """ Task instance started callback """
4609         self.log.debug("Got instance started callback")
4610
4611     @asyncio.coroutine
4612     def init(self):
4613         """ Task init callback """
4614         self.log.debug("Got instance started callback")
4615
4616         self.log.debug("creating project handler")
4617         self.project_handler = ProjectHandler(self, NsmProject)
4618         self.project_handler.register()
4619
4620
4621
4622     @asyncio.coroutine
4623     def run(self):
4624         """ Task run callback """
4625         pass
4626
4627     @asyncio.coroutine
4628     def on_dts_state_change(self, state):
4629         """Take action according to current dts state to transition
4630         application into the corresponding application state
4631
4632         Arguments
4633             state - current dts state
4634         """
4635         switch = {
4636             rwdts.State.INIT: rwdts.State.REGN_COMPLETE,
4637             rwdts.State.CONFIG: rwdts.State.RUN,
4638         }
4639
4640         handlers = {
4641             rwdts.State.INIT: self.init,
4642             rwdts.State.RUN: self.run,
4643         }
4644
4645         # Transition application to next state
4646         handler = handlers.get(state, None)
4647         if handler is not None:
4648             yield from handler()
4649
4650         # Transition dts to next state
4651         next_state = switch.get(state, None)
4652         if next_state is not None:
4653             self.log.debug("Changing state to %s", next_state)
4654             self._dts.handle.set_state(next_state)