rwlaunchpad/plugins/rwvnfm/rift/tasklets/rwvnfmtasklet/rwvnfmtasklet.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 import asyncio
  18 import collections
  19 import enum
  20 import logging
  21 import uuid
  22 import time
  23 import os.path
  24 import re
  25 import shutil
  26 import sys
  27
  28 import gi
  29 gi.require_version('RwDts', '1.0')
  30 gi.require_version('RwVnfrYang', '1.0')
  31 gi.require_version('RwVnfmYang', '1.0')
  32 gi.require_version('RwVlrYang', '1.0')
  33 gi.require_version('RwManifestYang', '1.0')
  34 gi.require_version('RwBaseYang', '1.0')
  35 gi.require_version('RwResourceMgrYang', '1.0')
  36
  37 from gi.repository import (
  38     RwDts as rwdts,
  39     RwVnfrYang,
  40     RwVnfmYang,
  41     RwVlrYang,
  42     VnfrYang,
  43     RwManifestYang,
  44     RwBaseYang,
  45     RwResourceMgrYang,
  46     ProtobufC,
  47 )
  48
  49 import rift.tasklets
  50 import rift.package.store
  51 import rift.package.cloud_init
  52
  53
  54 class VMResourceError(Exception):
  55     """ VM resource Error"""
  56     pass
  57
  58
  59 class VnfRecordError(Exception):
  60     """ VNF record instatiation failed"""
  61     pass
  62
  63
  64 class VduRecordError(Exception):
  65     """ VDU record instatiation failed"""
  66     pass
  67
  68
  69 class NotImplemented(Exception):
  70     """Not implemented """
  71     pass
  72
  73
  74 class VnfrRecordExistsError(Exception):
  75     """VNFR record already exist with the same VNFR id"""
  76     pass
  77
  78
  79 class InternalVirtualLinkRecordError(Exception):
  80     """Internal virtual link record error"""
  81     pass
  82
  83
  84 class VDUImageNotFound(Exception):
  85     """VDU Image not found error"""
  86     pass
  87
  88
  89 class VirtualDeploymentUnitRecordError(Exception):
  90     """VDU Instantiation failed"""
  91     pass
  92
  93
  94 class VMNotReadyError(Exception):
  95     """ VM Not yet received from resource manager """
  96     pass
  97
  98
  99 class VDURecordNotFound(Exception):
 100     """ Could not find a VDU record """
 101     pass
 102
 103
 104 class VirtualNetworkFunctionRecordDescNotFound(Exception):
 105     """ Cannot find Virtual Network Function Record Descriptor """
 106     pass
 107
 108
 109 class VirtualNetworkFunctionDescriptorError(Exception):
 110     """ Virtual Network Function Record Descriptor Error """
 111     pass
 112
 113
 114 class VirtualNetworkFunctionDescriptorNotFound(Exception):
 115     """ Virtual Network Function Record Descriptor Not Found """
 116     pass
 117
 118
 119 class VirtualNetworkFunctionRecordNotFound(Exception):
 120     """ Virtual Network Function Record Not Found """
 121     pass
 122
 123
 124 class VirtualNetworkFunctionDescriptorRefCountExists(Exception):
 125     """ Virtual Network Funtion Descriptor reference count exists """
 126     pass
 127
 128
 129 class VnfrInstantiationFailed(Exception):
 130     """ Virtual Network Funtion Instantiation failed"""
 131     pass
 132
 133
 134 class VNFMPlacementGroupError(Exception):
 135     pass
 136
 137 class VirtualNetworkFunctionRecordState(enum.Enum):
 138     """ VNFR state """
 139     INIT = 1
 140     VL_INIT_PHASE = 2
 141     VM_INIT_PHASE = 3
 142     READY = 4
 143     TERMINATE = 5
 144     VL_TERMINATE_PHASE = 6
 145     VDU_TERMINATE_PHASE = 7
 146     TERMINATED = 7
 147     FAILED = 10
 148
 149
 150 class VDURecordState(enum.Enum):
 151     """VDU record state """
 152     INIT = 1
 153     INSTANTIATING = 2
 154     RESOURCE_ALLOC_PENDING = 3
 155     READY = 4
 156     TERMINATING = 5
 157     TERMINATED = 6
 158     FAILED = 10
 159
 160
 161 class VcsComponent(object):
 162     """ VCS Component within the VNF descriptor """
 163     def __init__(self, dts, log, loop, cluster_name, vcs_handler, component, mangled_name):
 164         self._dts = dts
 165         self._log = log
 166         self._loop = loop
 167         self._component = component
 168         self._cluster_name = cluster_name
 169         self._vcs_handler = vcs_handler
 170         self._mangled_name = mangled_name
 171
 172     @staticmethod
 173     def mangle_name(component_name, vnf_name, vnfd_id):
 174         """ mangled  component name """
 175         return vnf_name + ":" + component_name + ":" + vnfd_id
 176
 177     @property
 178     def name(self):
 179         """ name of this component"""
 180         return self._mangled_name
 181
 182     @property
 183     def path(self):
 184         """ The path for this object """
 185         return("D,/rw-manifest:manifest" +
 186                "/rw-manifest:operational-inventory" +
 187                "/rw-manifest:component" +
 188                "[rw-manifest:component-name = '{}']").format(self.name)
 189
 190     @property
 191     def instance_xpath(self):
 192         """ The path for this object """
 193         return("D,/rw-base:vcs" +
 194                "/instances" +
 195                "/instance" +
 196                "[instance-name = '{}']".format(self._cluster_name))
 197
 198     @property
 199     def start_comp_xpath(self):
 200         """ start component xpath """
 201         return (self.instance_xpath +
 202                 "/child-n[instance-name = 'START-REQ']")
 203
 204     def get_start_comp_msg(self, ip_address):
 205         """ start this component """
 206         start_msg = RwBaseYang.VcsInstance_Instance_ChildN()
 207         start_msg.instance_name = 'START-REQ'
 208         start_msg.component_name = self.name
 209         start_msg.admin_command = "START"
 210         start_msg.ip_address = ip_address
 211
 212         return start_msg
 213
 214     @property
 215     def msg(self):
 216         """ Returns the message for this vcs component"""
 217
 218         vcs_comp_dict = self._component.as_dict()
 219
 220         def mangle_comp_names(comp_dict):
 221             """ mangle component name  with VNF name, id"""
 222             for key, val in comp_dict.items():
 223                 if isinstance(val, dict):
 224                     comp_dict[key] = mangle_comp_names(val)
 225                 elif isinstance(val, list):
 226                     i = 0
 227                     for ent in val:
 228                         if isinstance(ent, dict):
 229                             val[i] = mangle_comp_names(ent)
 230                         else:
 231                             val[i] = ent
 232                         i += 1
 233                 elif key == "component_name":
 234                     comp_dict[key] = VcsComponent.mangle_name(val,
 235                                                               self._vnfd_name,
 236                                                               self._vnfd_id)
 237             return comp_dict
 238
 239         mangled_dict = mangle_comp_names(vcs_comp_dict)
 240         msg = RwManifestYang.OpInventory_Component.from_dict(mangled_dict)
 241         return msg
 242
 243     @asyncio.coroutine
 244     def publish(self, xact):
 245         """ Publishes the VCS component """
 246         self._log.debug("Publishing the VcsComponent %s, path = %s comp = %s",
 247                         self.name, self.path, self.msg)
 248         yield from self._vcs_handler.publish(xact, self.path, self.msg)
 249
 250     @asyncio.coroutine
 251     def start(self, xact, parent, ip_addr=None):
 252         """ Starts this VCS component """
 253         # ATTN RV - replace with block add
 254         start_msg = self.get_start_comp_msg(ip_addr)
 255         self._log.debug("starting component %s %s",
 256                         self.start_comp_xpath, start_msg)
 257         yield from self._dts.query_create(self.start_comp_xpath,
 258                                           0,
 259                                           start_msg)
 260         self._log.debug("started component %s, %s",
 261                         self.start_comp_xpath, start_msg)
 262
 263
 264 class VirtualDeploymentUnitRecord(object):
 265     """  Virtual Deployment Unit Record """
 266     def __init__(self,
 267                  dts,
 268                  log,
 269                  loop,
 270                  vdud,
 271                  vnfr,
 272                  mgmt_intf,
 273                  cloud_account_name,
 274                  vnfd_package_store,
 275                  vdur_id=None,
 276                  placement_groups=[]):
 277         self._dts = dts
 278         self._log = log
 279         self._loop = loop
 280         self._vdud = vdud
 281         self._vnfr = vnfr
 282         self._mgmt_intf = mgmt_intf
 283         self._cloud_account_name = cloud_account_name
 284         self._vnfd_package_store = vnfd_package_store
 285
 286         self._vdur_id = vdur_id or str(uuid.uuid4())
 287         self._int_intf = []
 288         self._ext_intf = []
 289         self._state = VDURecordState.INIT
 290         self._state_failed_reason = None
 291         self._request_id = str(uuid.uuid4())
 292         self._name = vnfr.name + "__" + vdud.id
 293         self._placement_groups = placement_groups
 294         self._rm_regh = None
 295         self._vm_resp = None
 296         self._vdud_cloud_init = None
 297         self._vdur_console_handler = VnfrConsoleOperdataDtsHandler(dts, log, loop, self._vnfr._vnfm, self._vnfr.vnfr_id, self._vdur_id,self.vdu_id)
 298
 299     @asyncio.coroutine
 300     def vdu_opdata_register(self):
 301         yield from self._vdur_console_handler.register()
 302
 303     def cp_ip_addr(self, cp_name):
 304         """ Find ip address by connection point name """
 305         if self._vm_resp is not None:
 306             for conn_point in self._vm_resp.connection_points:
 307                 if conn_point.name == cp_name:
 308                     return conn_point.ip_address
 309         return "0.0.0.0"
 310
 311     def cp_mac_addr(self, cp_name):
 312         """ Find mac address by connection point name """
 313         if self._vm_resp is not None:
 314             for conn_point in self._vm_resp.connection_points:
 315                 if conn_point.name == cp_name:
 316                     return conn_point.mac_addr
 317         return "00:00:00:00:00:00"
 318
 319     def cp_id(self, cp_name):
 320         """ Find connection point id  by connection point name """
 321         if self._vm_resp is not None:
 322             for conn_point in self._vm_resp.connection_points:
 323                 if conn_point.name == cp_name:
 324                     return conn_point.connection_point_id
 325         return ''
 326
 327     @property
 328     def vdu_id(self):
 329         return self._vdud.id
 330
 331     @property
 332     def vm_resp(self):
 333         return self._vm_resp
 334
 335     @property
 336     def name(self):
 337         """ Return this VDUR's name """
 338         return self._name
 339
 340     @property
 341     def cloud_account_name(self):
 342         """ Cloud account this VDU should be created in """
 343         return self._cloud_account_name
 344
 345     @property
 346     def image_name(self):
 347         """ name that should be used to lookup the image on the CMP """
 348         return os.path.basename(self._vdud.image)
 349
 350     @property
 351     def image_checksum(self):
 352         """ name that should be used to lookup the image on the CMP """
 353         return self._vdud.image_checksum if self._vdud.has_field("image_checksum") else None
 354
 355     @property
 356     def management_ip(self):
 357         if not self.active:
 358             return None
 359         return self._vm_resp.public_ip if self._vm_resp.has_field('public_ip') else self._vm_resp.management_ip
 360
 361     @property
 362     def vm_management_ip(self):
 363         if not self.active:
 364             return None
 365         return self._vm_resp.management_ip
 366
 367     @property
 368     def operational_status(self):
 369         """ Operational status of this VDU"""
 370         op_stats_dict = {"INIT": "init",
 371                          "INSTANTIATING": "vm_init_phase",
 372                          "RESOURCE_ALLOC_PENDING": "vm_alloc_pending",
 373                          "READY": "running",
 374                          "FAILED": "failed",
 375                          "TERMINATING": "terminated",
 376                          "TERMINATED": "terminated",
 377                          }
 378         return op_stats_dict[self._state.name]
 379
 380     @property
 381     def msg(self):
 382         """ VDU message """
 383         vdu_fields = ["vm_flavor",
 384                       "guest_epa",
 385                       "vswitch_epa",
 386                       "hypervisor_epa",
 387                       "host_epa",
 388                       "name"]
 389         vdu_copy_dict = {k: v for k, v in
 390                          self._vdud.as_dict().items() if k in vdu_fields}
 391         vdur_dict = {"id": self._vdur_id,
 392                      "vdu_id_ref": self._vdud.id,
 393                      "operational_status": self.operational_status,
 394                      "operational_status_details": self._state_failed_reason,
 395                      }
 396         if self.vm_resp is not None:
 397             vdur_dict.update({"vim_id": self.vm_resp.vdu_id,
 398                               "flavor_id": self.vm_resp.flavor_id,
 399                               "image_id": self.vm_resp.image_id,
 400                               })
 401
 402         if self.management_ip is not None:
 403             vdur_dict["management_ip"] = self.management_ip
 404
 405         if self.vm_management_ip is not None:
 406             vdur_dict["vm_management_ip"] = self.vm_management_ip
 407
 408         vdur_dict.update(vdu_copy_dict)
 409
 410         icp_list = []
 411         ii_list = []
 412
 413         for intf, cp_id, vlr in self._int_intf:
 414             cp = self.find_internal_cp_by_cp_id(cp_id)
 415
 416             icp_list.append({"name": cp.name,
 417                              "id": cp.id,
 418                              "type_yang": "VPORT",
 419                              "ip_address": self.cp_ip_addr(cp.id),
 420                              "mac_address": self.cp_mac_addr(cp.id)})
 421
 422             ii_list.append({"name": intf.name,
 423                             "vdur_internal_connection_point_ref": cp.id,
 424                             "virtual_interface": {}})
 425
 426         vdur_dict["internal_connection_point"] = icp_list
 427         self._log.debug("internal_connection_point:%s", vdur_dict["internal_connection_point"])
 428         vdur_dict["internal_interface"] = ii_list
 429
 430         ei_list = []
 431         for intf, cp, vlr in self._ext_intf:
 432             ei_list.append({"name": cp,
 433                             "vnfd_connection_point_ref": cp,
 434                             "virtual_interface": {}})
 435             self._vnfr.update_cp(cp,
 436                                  self.cp_ip_addr(cp),
 437                                  self.cp_mac_addr(cp),
 438                                  self.cp_id(cp))
 439
 440         vdur_dict["external_interface"] = ei_list
 441
 442         placement_groups = []
 443         for group in self._placement_groups:
 444             placement_groups.append(group.as_dict())
 445
 446         vdur_dict['placement_groups_info'] = placement_groups
 447         return RwVnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_Vdur.from_dict(vdur_dict)
 448
 449     @property
 450     def resmgr_path(self):
 451         """ path for resource-mgr"""
 452         return ("D,/rw-resource-mgr:resource-mgmt" +
 453                 "/vdu-event" +
 454                 "/vdu-event-data[event-id='{}']".format(self._request_id))
 455
 456     @property
 457     def vm_flavor_msg(self):
 458         """ VM flavor message """
 459         flavor = self._vdud.vm_flavor.__class__()
 460         flavor.copy_from(self._vdud.vm_flavor)
 461
 462         return flavor
 463
 464     @property
 465     def vdud_cloud_init(self):
 466         """ Return the cloud-init contents for the VDU """
 467         if self._vdud_cloud_init is None:
 468             self._vdud_cloud_init = self.cloud_init()
 469
 470         return self._vdud_cloud_init
 471
 472     def cloud_init(self):
 473         """ Populate cloud_init with cloud-config script from
 474             either the inline contents or from the file provided
 475         """
 476         if self._vdud.cloud_init is not None:
 477             self._log.debug("cloud_init script provided inline %s", self._vdud.cloud_init)
 478             return self._vdud.cloud_init
 479         elif self._vdud.cloud_init_file is not None:
 480             # Get cloud-init script contents from the file provided in the cloud_init_file param
 481             self._log.debug("cloud_init script provided in file %s", self._vdud.cloud_init_file)
 482             filename = self._vdud.cloud_init_file
 483             self._vnfd_package_store.refresh()
 484             stored_package = self._vnfd_package_store.get_package(self._vnfr.vnfd_id)
 485             cloud_init_extractor = rift.package.cloud_init.PackageCloudInitExtractor(self._log)
 486             try:
 487                 return cloud_init_extractor.read_script(stored_package, filename)
 488             except rift.package.cloud_init.CloudInitExtractionError as e:
 489                 raise VirtualDeploymentUnitRecordError(e)
 490         else:
 491             self._log.debug("VDU Instantiation: cloud-init script not provided")
 492
 493     def process_openstack_placement_group_construct(self, vm_create_msg_dict):
 494         host_aggregates = []
 495         availability_zones = []
 496         server_groups = []
 497         for group in self._placement_groups:
 498             if group.has_field('host_aggregate'):
 499                 for aggregate in group.host_aggregate:
 500                     host_aggregates.append(aggregate.as_dict())
 501             if group.has_field('availability_zone'):
 502                 availability_zones.append(group.availability_zone.as_dict())
 503             if group.has_field('server_group'):
 504                 server_groups.append(group.server_group.as_dict())
 505
 506         if availability_zones:
 507             if len(availability_zones) > 1:
 508                 self._log.error("Can not launch VDU: %s in multiple availability zones. Requested Zones: %s", self.name, availability_zones)
 509                 raise VNFMPlacementGroupError("Can not launch VDU: {} in multiple availability zones. Requsted Zones".format(self.name, availability_zones))
 510             else:
 511                 vm_create_msg_dict['availability_zone'] = availability_zones[0]
 512
 513         if server_groups:
 514             if len(server_groups) > 1:
 515                 self._log.error("Can not launch VDU: %s in multiple Server Group. Requested Groups: %s", self.name, server_groups)
 516                 raise VNFMPlacementGroupError("Can not launch VDU: {} in multiple Server Groups. Requsted Groups".format(self.name, server_groups))
 517             else:
 518                 vm_create_msg_dict['server_group'] = server_groups[0]
 519
 520         if host_aggregates:
 521             vm_create_msg_dict['host_aggregate'] = host_aggregates
 522
 523         return
 524
 525     def process_placement_groups(self, vm_create_msg_dict):
 526         """Process the placement_groups and fill resource-mgr request"""
 527         if not self._placement_groups:
 528             return
 529
 530         cloud_set = set([group.cloud_type for group in self._placement_groups])
 531         assert len(cloud_set) == 1
 532         cloud_type = cloud_set.pop()
 533
 534         if cloud_type == 'openstack':
 535             self.process_openstack_placement_group_construct(vm_create_msg_dict)
 536
 537         else:
 538             self._log.info("Ignoring placement group with cloud construct for cloud-type: %s", cloud_type)
 539         return
 540
 541     def resmgr_msg(self, config=None):
 542         vdu_fields = ["vm_flavor",
 543                       "guest_epa",
 544                       "vswitch_epa",
 545                       "hypervisor_epa",
 546                       "host_epa"]
 547
 548         self._log.debug("Creating params based on VDUD: %s", self._vdud)
 549         vdu_copy_dict = {k: v for k, v in self._vdud.as_dict().items() if k in vdu_fields}
 550
 551         vm_create_msg_dict = {
 552                 "name": self.name,
 553                 "image_name": self.image_name,
 554                 }
 555
 556         if self.image_checksum is not None:
 557             vm_create_msg_dict["image_checksum"] = self.image_checksum
 558
 559         vm_create_msg_dict["allocate_public_address"] = self._mgmt_intf
 560         if self._vdud.has_field('mgmt_vpci'):
 561             vm_create_msg_dict["mgmt_vpci"] = self._vdud.mgmt_vpci
 562
 563         self._log.debug("VDUD: %s", self._vdud)
 564         if config is not None:
 565             vm_create_msg_dict['vdu_init'] = {'userdata': config}
 566
 567         cp_list = []
 568         for intf, cp, vlr in self._ext_intf:
 569             cp_info = {"name": cp,
 570                        "virtual_link_id": vlr.network_id,
 571                        "type_yang": intf.virtual_interface.type_yang}
 572
 573             if (intf.virtual_interface.has_field('vpci') and
 574                     intf.virtual_interface.vpci is not None):
 575                 cp_info["vpci"] =  intf.virtual_interface.vpci
 576
 577             if (vlr.has_field('ip_profile_params')) and (vlr.ip_profile_params.has_field('security_group')):
 578                 cp_info['security_group'] = vlr.ip_profile_params.security_group
 579
 580             cp_list.append(cp_info)
 581
 582         for intf, cp, vlr in self._int_intf:
 583             if (intf.virtual_interface.has_field('vpci') and
 584                     intf.virtual_interface.vpci is not None):
 585                 cp_list.append({"name": cp,
 586                                 "virtual_link_id": vlr.network_id,
 587                                 "type_yang": intf.virtual_interface.type_yang,
 588                                 "vpci": intf.virtual_interface.vpci})
 589             else:
 590                 cp_list.append({"name": cp,
 591                                 "virtual_link_id": vlr.network_id,
 592                                 "type_yang": intf.virtual_interface.type_yang})
 593
 594         vm_create_msg_dict["connection_points"] = cp_list
 595         vm_create_msg_dict.update(vdu_copy_dict)
 596
 597         self.process_placement_groups(vm_create_msg_dict)
 598
 599         msg = RwResourceMgrYang.VDUEventData()
 600         msg.event_id = self._request_id
 601         msg.cloud_account = self.cloud_account_name
 602         msg.request_info.from_dict(vm_create_msg_dict)
 603         return msg
 604
 605     @asyncio.coroutine
 606     def terminate(self, xact):
 607         """ Delete resource in VIM """
 608         if self._state != VDURecordState.READY and self._state != VDURecordState.FAILED:
 609             self._log.warning("VDU terminate in not ready state - Ignoring request")
 610             return
 611
 612         self._state = VDURecordState.TERMINATING
 613         if self._vm_resp is not None:
 614             try:
 615                 with self._dts.transaction() as new_xact:
 616                     yield from self.delete_resource(new_xact)
 617             except Exception:
 618                 self._log.exception("Caught exception while deleting VDU %s", self.vdu_id)
 619
 620         if self._rm_regh is not None:
 621             self._log.debug("Deregistering resource manager registration handle")
 622             self._rm_regh.deregister()
 623             self._rm_regh = None
 624
 625         if self._vdur_console_handler is not None:
 626             self._log.error("Deregistering vnfr vdur registration handle")
 627             self._vdur_console_handler._regh.deregister()
 628             self._vdur_console_handler._regh = None
 629
 630         self._state = VDURecordState.TERMINATED
 631
 632     def find_internal_cp_by_cp_id(self, cp_id):
 633         """ Find the CP corresponding to the connection point id"""
 634         cp = None
 635
 636         self._log.debug("find_internal_cp_by_cp_id(%s) called",
 637                         cp_id)
 638
 639         for int_cp in self._vdud.internal_connection_point:
 640             self._log.debug("Checking for int cp %s in internal connection points",
 641                             int_cp.id)
 642             if int_cp.id == cp_id:
 643                 cp = int_cp
 644                 break
 645
 646         if cp is None:
 647             self._log.debug("Failed to find cp %s in internal connection points",
 648                             cp_id)
 649             msg = "Failed to find cp %s in internal connection points" % cp_id
 650             raise VduRecordError(msg)
 651
 652         # return the VLR associated with the connection point
 653         return cp
 654
 655     @asyncio.coroutine
 656     def create_resource(self, xact, vnfr, config=None):
 657         """ Request resource from ResourceMgr """
 658         def find_cp_by_name(cp_name):
 659             """ Find a connection point by name """
 660             cp = None
 661             self._log.debug("find_cp_by_name(%s) called", cp_name)
 662             for ext_cp in vnfr._cprs:
 663                 self._log.debug("Checking ext cp (%s) called", ext_cp.name)
 664                 if ext_cp.name == cp_name:
 665                     cp = ext_cp
 666                     break
 667             if cp is None:
 668                 self._log.debug("Failed to find cp %s in external connection points",
 669                                 cp_name)
 670             return cp
 671
 672         def find_internal_vlr_by_cp_name(cp_name):
 673             """ Find the VLR corresponding to the connection point name"""
 674             cp = None
 675
 676             self._log.debug("find_internal_vlr_by_cp_name(%s) called",
 677                             cp_name)
 678
 679             for int_cp in self._vdud.internal_connection_point:
 680                 self._log.debug("Checking for int cp %s in internal connection points",
 681                                 int_cp.id)
 682                 if int_cp.id == cp_name:
 683                     cp = int_cp
 684                     break
 685
 686             if cp is None:
 687                 self._log.debug("Failed to find cp %s in internal connection points",
 688                                 cp_name)
 689                 msg = "Failed to find cp %s in internal connection points" % cp_name
 690                 raise VduRecordError(msg)
 691
 692             # return the VLR associated with the connection point
 693             return vnfr.find_vlr_by_cp(cp_name)
 694
 695         block = xact.block_create()
 696
 697         self._log.debug("Executing vm request id: %s, action: create",
 698                         self._request_id)
 699
 700         # Resolve the networks associated external interfaces
 701         for ext_intf in self._vdud.external_interface:
 702             self._log.debug("Resolving external interface name [%s], cp[%s]",
 703                             ext_intf.name, ext_intf.vnfd_connection_point_ref)
 704             cp = find_cp_by_name(ext_intf.vnfd_connection_point_ref)
 705             if cp is None:
 706                 self._log.debug("Failed to find connection point - %s",
 707                                 ext_intf.vnfd_connection_point_ref)
 708                 continue
 709             self._log.debug("Connection point name [%s], type[%s]",
 710                             cp.name, cp.type_yang)
 711
 712             vlr = vnfr.ext_vlr_by_id(cp.vlr_ref)
 713
 714             etuple = (ext_intf, cp.name, vlr)
 715             self._ext_intf.append(etuple)
 716
 717             self._log.debug("Created external interface tuple  : %s", etuple)
 718
 719         # Resolve the networks associated internal interfaces
 720         for intf in self._vdud.internal_interface:
 721             cp_id = intf.vdu_internal_connection_point_ref
 722             self._log.debug("Resolving internal interface name [%s], cp[%s]",
 723                             intf.name, cp_id)
 724
 725             try:
 726                 vlr = find_internal_vlr_by_cp_name(cp_id)
 727             except Exception as e:
 728                 self._log.debug("Failed to find cp %s in internal VLR list", cp_id)
 729                 msg = "Failed to find cp %s in internal VLR list, e = %s" % (cp_id, e)
 730                 raise VduRecordError(msg)
 731
 732             ituple = (intf, cp_id, vlr)
 733             self._int_intf.append(ituple)
 734
 735             self._log.debug("Created internal interface tuple  : %s", ituple)
 736
 737         resmgr_path = self.resmgr_path
 738         resmgr_msg = self.resmgr_msg(config)
 739
 740         self._log.debug("Creating new VM request at: %s, params: %s", resmgr_path, resmgr_msg)
 741         block.add_query_create(resmgr_path, resmgr_msg)
 742
 743         res_iter = yield from block.execute(now=True)
 744
 745         resp = None
 746
 747         for i in res_iter:
 748             r = yield from i
 749             resp = r.result
 750
 751         if resp is None or not (resp.has_field('resource_info') and resp.resource_info.has_field('resource_state')):
 752             raise VMResourceError("Did not get a vm resource response (resp: %s)", resp)
 753         self._log.debug("Got vm request response: %s", resp.resource_info)
 754         return resp.resource_info
 755
 756     @asyncio.coroutine
 757     def delete_resource(self, xact):
 758         block = xact.block_create()
 759
 760         self._log.debug("Executing vm request id: %s, action: delete",
 761                         self._request_id)
 762
 763         block.add_query_delete(self.resmgr_path)
 764
 765         yield from block.execute(flags=0, now=True)
 766
 767     @asyncio.coroutine
 768     def read_resource(self, xact):
 769         block = xact.block_create()
 770
 771         self._log.debug("Executing vm request id: %s, action: delete",
 772                         self._request_id)
 773
 774         block.add_query_read(self.resmgr_path)
 775
 776         res_iter = yield from block.execute(flags=0, now=True)
 777         for i in res_iter:
 778             r = yield from i
 779             resp = r.result
 780
 781         if resp is None or not (resp.has_field('resource_info') and resp.resource_info.has_field('resource_state')):
 782             raise VMResourceError("Did not get a vm resource response (resp: %s)", resp)
 783         self._log.debug("Got vm request response: %s", resp.resource_info)
 784         #self._vm_resp = resp.resource_info
 785         return resp.resource_info
 786
 787
 788     @asyncio.coroutine
 789     def start_component(self):
 790         """ This VDUR is active """
 791         self._log.debug("Starting component %s for  vdud %s vdur %s",
 792                         self._vdud.vcs_component_ref,
 793                         self._vdud,
 794                         self._vdur_id)
 795         yield from self._vnfr.start_component(self._vdud.vcs_component_ref,
 796                                               self.vm_resp.management_ip)
 797
 798     @property
 799     def active(self):
 800         """ Is this VDU active """
 801         return True if self._state is VDURecordState.READY else False
 802
 803     @asyncio.coroutine
 804     def instantiation_failed(self, failed_reason=None):
 805         """ VDU instantiation failed """
 806         self._log.debug("VDU %s instantiation failed ", self._vdur_id)
 807         self._state = VDURecordState.FAILED
 808         self._state_failed_reason = failed_reason
 809         yield from self._vnfr.instantiation_failed(failed_reason)
 810
 811     @asyncio.coroutine
 812     def vdu_is_active(self):
 813         """ This VDU is active"""
 814         if self.active:
 815             self._log.warning("VDU %s was already marked as active", self._vdur_id)
 816             return
 817
 818         self._log.debug("VDUR id %s in VNFR %s is active", self._vdur_id, self._vnfr.vnfr_id)
 819
 820         if self._vdud.vcs_component_ref is not None:
 821             yield from self.start_component()
 822
 823         self._state = VDURecordState.READY
 824
 825         if self._vnfr.all_vdus_active():
 826             self._log.debug("Inside vdu_is_active. VNFR is READY. Info: %s", self._vnfr)
 827             yield from self._vnfr.is_ready()
 828
 829     @asyncio.coroutine
 830     def instantiate(self, xact, vnfr, config=None):
 831         """ Instantiate this VDU """
 832         self._state = VDURecordState.INSTANTIATING
 833
 834         @asyncio.coroutine
 835         def on_prepare(xact_info, query_action, ks_path, msg):
 836             """ This VDUR is active """
 837             self._log.debug("Received VDUR instantiate on_prepare (%s:%s:%s)",
 838                             query_action,
 839                             ks_path,
 840                             msg)
 841
 842             if (query_action == rwdts.QueryAction.UPDATE or
 843                     query_action == rwdts.QueryAction.CREATE):
 844                 self._vm_resp = msg
 845
 846                 if msg.resource_state == "active":
 847                     # Move this VDU to ready state
 848                     yield from self.vdu_is_active()
 849                 elif msg.resource_state == "failed":
 850                     yield from self.instantiation_failed(msg.resource_errors)
 851             elif query_action == rwdts.QueryAction.DELETE:
 852                 self._log.debug("DELETE action in on_prepare for VDUR instantiation, ignoring")
 853             else:
 854                 raise NotImplementedError(
 855                     "%s action on VirtualDeployementUnitRecord not supported",
 856                     query_action)
 857
 858             xact_info.respond_xpath(rwdts.XactRspCode.ACK)
 859
 860         try:
 861             reg_event = asyncio.Event(loop=self._loop)
 862
 863             @asyncio.coroutine
 864             def on_ready(regh, status):
 865                 reg_event.set()
 866
 867             handler = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_prepare, on_ready=on_ready)
 868             self._rm_regh = yield from self._dts.register(self.resmgr_path + '/resource-info',
 869                                                           flags=rwdts.Flag.SUBSCRIBER,
 870                                                           handler=handler)
 871             yield from reg_event.wait()
 872
 873             vm_resp = yield from self.create_resource(xact, vnfr, config)
 874             self._vm_resp = vm_resp
 875
 876             self._state = VDURecordState.RESOURCE_ALLOC_PENDING
 877             self._log.debug("Requested VM from resource manager response %s",
 878                             vm_resp)
 879             if vm_resp.resource_state == "active":
 880                 self._log.debug("Resourcemgr responded wih an active vm resp %s",
 881                                 vm_resp)
 882                 yield from self.vdu_is_active()
 883                 self._state = VDURecordState.READY
 884             elif (vm_resp.resource_state == "pending" or
 885                   vm_resp.resource_state == "inactive"):
 886                 self._log.debug("Resourcemgr responded wih a pending vm resp %s",
 887                                 vm_resp)
 888                 # handler = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_prepare)
 889                 # self._rm_regh = yield from self._dts.register(self.resmgr_path + '/resource-info',
 890                 #                                              flags=rwdts.Flag.SUBSCRIBER,
 891                 #                                              handler=handler)
 892             else:
 893                 self._log.debug("Resourcemgr responded wih an error vm resp %s",
 894                                 vm_resp)
 895                 raise VirtualDeploymentUnitRecordError(
 896                     "Failed VDUR instantiation %s " % vm_resp)
 897
 898         except Exception as e:
 899             import traceback
 900             traceback.print_exc()
 901             self._log.exception(e)
 902             self._log.error("Instantiation of VDU record failed: %s", str(e))
 903             self._state = VDURecordState.FAILED
 904             yield from self.instantiation_failed(str(e))
 905
 906
 907 class VlRecordState(enum.Enum):
 908     """ VL Record State """
 909     INIT = 101
 910     INSTANTIATION_PENDING = 102
 911     ACTIVE = 103
 912     TERMINATE_PENDING = 104
 913     TERMINATED = 105
 914     FAILED = 106
 915
 916
 917 class InternalVirtualLinkRecord(object):
 918     """ Internal Virtual Link record """
 919     def __init__(self, dts, log, loop, ivld_msg, vnfr_name, cloud_account_name):
 920         self._dts = dts
 921         self._log = log
 922         self._loop = loop
 923         self._ivld_msg = ivld_msg
 924         self._vnfr_name = vnfr_name
 925         self._cloud_account_name = cloud_account_name
 926
 927         self._vlr_req = self.create_vlr()
 928         self._vlr = None
 929         self._state = VlRecordState.INIT
 930
 931     @property
 932     def vlr_id(self):
 933         """ Find VLR by id """
 934         return self._vlr_req.id
 935
 936     @property
 937     def name(self):
 938         """ Name of this VL """
 939         return self._vnfr_name + "." + self._ivld_msg.name
 940
 941     @property
 942     def network_id(self):
 943         """ Find VLR by id """
 944         return self._vlr.network_id if self._vlr else None
 945
 946     def vlr_path(self):
 947         """ VLR path for this VLR instance"""
 948         return "D,/vlr:vlr-catalog/vlr:vlr[vlr:id = '{}']".format(self.vlr_id)
 949
 950     def create_vlr(self):
 951         """ Create the VLR record which will be instantiated """
 952
 953         vld_fields = ["short_name",
 954                       "vendor",
 955                       "description",
 956                       "version",
 957                       "type_yang",
 958                       "provider_network"]
 959
 960         vld_copy_dict = {k: v for k, v in self._ivld_msg.as_dict().items() if k in vld_fields}
 961
 962         vlr_dict = {"id": str(uuid.uuid4()),
 963                     "name": self.name,
 964                     "cloud_account": self._cloud_account_name,
 965                     }
 966         vlr_dict.update(vld_copy_dict)
 967
 968         vlr = RwVlrYang.YangData_Vlr_VlrCatalog_Vlr.from_dict(vlr_dict)
 969         return vlr
 970
 971     @asyncio.coroutine
 972     def instantiate(self, xact, restart_mode=False):
 973         """ Instantiate VL """
 974
 975         @asyncio.coroutine
 976         def instantiate_vlr():
 977             """ Instantiate VLR"""
 978             self._log.debug("Create VL with xpath %s and vlr %s",
 979                             self.vlr_path(), self._vlr_req)
 980
 981             with self._dts.transaction(flags=0) as xact:
 982                 block = xact.block_create()
 983                 block.add_query_create(xpath=self.vlr_path(), msg=self._vlr_req)
 984                 self._log.debug("Executing VL create path:%s msg:%s",
 985                                 self.vlr_path(), self._vlr_req)
 986
 987                 res_iter = None
 988                 try:
 989                     res_iter = yield from block.execute()
 990                 except Exception:
 991                     self._state = VlRecordState.FAILED
 992                     self._log.exception("Caught exception while instantial VL")
 993                     raise
 994
 995                 for ent in res_iter:
 996                     res = yield from ent
 997                     self._vlr = res.result
 998
 999             if self._vlr.operational_status == 'failed':
1000                 self._log.debug("VL creation failed for vlr id %s", self._vlr.id)
1001                 self._state = VlRecordState.FAILED
1002                 raise VnfrInstantiationFailed("instantiation due to VL failure %s" % (self._vlr.id))
1003
1004             self._log.info("Created VL with xpath %s and vlr %s",
1005                            self.vlr_path(), self._vlr)
1006
1007         @asyncio.coroutine
1008         def get_vlr():
1009             """ Get the network id """
1010             res_iter = yield from self._dts.query_read(self.vlr_path(), rwdts.XactFlag.MERGE)
1011             vlr = None
1012             for ent in res_iter:
1013                 res = yield from ent
1014                 vlr = res.result
1015
1016             if vlr is None:
1017                 err = "Failed to get VLR for path  %s" % self.vlr_path()
1018                 self._log.warn(err)
1019                 raise InternalVirtualLinkRecordError(err)
1020             return vlr
1021
1022         self._state = VlRecordState.INSTANTIATION_PENDING
1023
1024         if restart_mode:
1025             vl = yield from get_vlr()
1026             if vl is None:
1027                 yield from instantiate_vlr()
1028         else:
1029             yield from instantiate_vlr()
1030
1031         self._state = VlRecordState.ACTIVE
1032
1033     def vlr_in_vns(self):
1034         """ Is there a VLR record in VNS """
1035         if (self._state == VlRecordState.ACTIVE or
1036                 self._state == VlRecordState.INSTANTIATION_PENDING or
1037                 self._state == VlRecordState.FAILED):
1038             return True
1039
1040         return False
1041
1042     @asyncio.coroutine
1043     def terminate(self, xact):
1044         """Terminate this VL """
1045         if not self.vlr_in_vns():
1046             self._log.debug("Ignoring terminate request for id %s in state %s",
1047                             self.vlr_id, self._state)
1048             return
1049
1050         self._log.debug("Terminating VL with path %s", self.vlr_path())
1051         self._state = VlRecordState.TERMINATE_PENDING
1052         block = xact.block_create()
1053         block.add_query_delete(self.vlr_path())
1054         yield from block.execute(flags=0, now=True)
1055         self._state = VlRecordState.TERMINATED
1056         self._log.debug("Terminated VL with path %s", self.vlr_path())
1057
1058
1059 class VirtualNetworkFunctionRecord(object):
1060     """ Virtual Network Function Record """
1061     def __init__(self, dts, log, loop, cluster_name, vnfm, vcs_handler, vnfr_msg):
1062         self._dts = dts
1063         self._log = log
1064         self._loop = loop
1065         self._cluster_name = cluster_name
1066         self._vnfr_msg = vnfr_msg
1067         self._vnfr_id = vnfr_msg.id
1068         self._vnfd_id = vnfr_msg.vnfd_ref
1069         self._vnfm = vnfm
1070         self._vcs_handler = vcs_handler
1071         self._vnfr = vnfr_msg
1072
1073         self._vnfd = None
1074         self._state = VirtualNetworkFunctionRecordState.INIT
1075         self._state_failed_reason = None
1076         self._ext_vlrs = {}  # The list of external virtual links
1077         self._vlrs = []  # The list of internal virtual links
1078         self._vdus = []  # The list of vdu
1079         self._vlr_by_cp = {}
1080         self._cprs = []
1081         self._inventory = {}
1082         self._create_time = int(time.time())
1083         self._vnf_mon = None
1084         self._config_status = vnfr_msg.config_status
1085         self._vnfd_package_store = rift.package.store.VnfdPackageFilesystemStore(self._log)
1086
1087     def _get_vdur_from_vdu_id(self, vdu_id):
1088         self._log.debug("Finding vdur for vdu_id %s", vdu_id)
1089         self._log.debug("Searching through vdus: %s", self._vdus)
1090         for vdu in self._vdus:
1091             self._log.debug("vdu_id: %s", vdu.vdu_id)
1092             if vdu.vdu_id == vdu_id:
1093                 return vdu
1094
1095         raise VDURecordNotFound("Could not find vdu record from id: %s", vdu_id)
1096
1097     @property
1098     def operational_status(self):
1099         """ Operational status of this VNFR """
1100         op_status_map = {"INIT": "init",
1101                          "VL_INIT_PHASE": "vl_init_phase",
1102                          "VM_INIT_PHASE": "vm_init_phase",
1103                          "READY": "running",
1104                          "TERMINATE": "terminate",
1105                          "VL_TERMINATE_PHASE": "vl_terminate_phase",
1106                          "VDU_TERMINATE_PHASE": "vm_terminate_phase",
1107                          "TERMINATED": "terminated",
1108                          "FAILED": "failed", }
1109         return op_status_map[self._state.name]
1110
1111     @property
1112     def vnfd_xpath(self):
1113         """ VNFD xpath associated with this VNFR """
1114         return("C,/vnfd:vnfd-catalog/"
1115                "vnfd:vnfd[vnfd:id = '{}']".format(self._vnfd_id))
1116
1117     @property
1118     def vnfd(self):
1119         """ VNFD for this VNFR """
1120         return self._vnfd
1121
1122     @property
1123     def vnf_name(self):
1124         """ VNFD name associated with this VNFR """
1125         return self.vnfd.name
1126
1127     @property
1128     def name(self):
1129         """ Name of this VNF in the record """
1130         return self._vnfr.name
1131
1132     @property
1133     def cloud_account_name(self):
1134         """ Name of the cloud account this VNFR is instantiated in """
1135         return self._vnfr.cloud_account
1136
1137     @property
1138     def vnfd_id(self):
1139         """ VNFD Id associated with this VNFR """
1140         return self.vnfd.id
1141
1142     @property
1143     def vnfr_id(self):
1144         """ VNFR Id associated with this VNFR """
1145         return self._vnfr_id
1146
1147     @property
1148     def member_vnf_index(self):
1149         """ Member VNF index associated with this VNFR """
1150         return self._vnfr.member_vnf_index_ref
1151
1152     @property
1153     def config_status(self):
1154         """ Config agent status for this VNFR """
1155         return self._config_status
1156
1157     def component_by_name(self, component_name):
1158         """ Find a component by name in the inventory list"""
1159         mangled_name = VcsComponent.mangle_name(component_name,
1160                                                 self.vnf_name,
1161                                                 self.vnfd_id)
1162         return self._inventory[mangled_name]
1163
1164
1165
1166     @asyncio.coroutine
1167     def get_nsr_config(self):
1168         ### Need access to NS instance configuration for runtime resolution.
1169         ### This shall be replaced when deployment flavors are implemented
1170         xpath = "C,/nsr:ns-instance-config"
1171         results = yield from self._dts.query_read(xpath, rwdts.XactFlag.MERGE)
1172
1173         for result in results:
1174             entry = yield from result
1175             ns_instance_config = entry.result
1176             for nsr in ns_instance_config.nsr:
1177                 if nsr.id == self._vnfr_msg.nsr_id_ref:
1178                     return nsr
1179         return None
1180
1181     @asyncio.coroutine
1182     def start_component(self, component_name, ip_addr):
1183         """ Start a component in the VNFR by name """
1184         comp = self.component_by_name(component_name)
1185         yield from comp.start(None, None, ip_addr)
1186
1187     def cp_ip_addr(self, cp_name):
1188         """ Get ip address for connection point """
1189         self._log.debug("cp_ip_addr()")
1190         for cp in self._cprs:
1191             if cp.name == cp_name and cp.ip_address is not None:
1192                 return cp.ip_address
1193         return "0.0.0.0"
1194
1195     def mgmt_intf_info(self):
1196         """ Get Management interface info for this VNFR """
1197         mgmt_intf_desc = self.vnfd.msg.mgmt_interface
1198         ip_addr = None
1199         if mgmt_intf_desc.has_field("cp"):
1200             ip_addr = self.cp_ip_addr(mgmt_intf_desc.cp)
1201         elif mgmt_intf_desc.has_field("vdu_id"):
1202             try:
1203                 vdur = self._get_vdur_from_vdu_id(mgmt_intf_desc.vdu_id)
1204                 ip_addr = vdur.management_ip
1205             except VDURecordNotFound:
1206                 self._log.debug("Did not find mgmt interface for vnfr id %s", self._vnfr_id)
1207                 ip_addr = None
1208         else:
1209             ip_addr = mgmt_intf_desc.ip_address
1210         port = mgmt_intf_desc.port
1211
1212         return ip_addr, port
1213
1214     @property
1215     def msg(self):
1216         """ Message associated with this VNFR """
1217         vnfd_fields = ["short_name", "vendor", "description", "version"]
1218         vnfd_copy_dict = {k: v for k, v in self.vnfd.msg.as_dict().items() if k in vnfd_fields}
1219
1220         mgmt_intf = VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_MgmtInterface()
1221         ip_address, port = self.mgmt_intf_info()
1222
1223         if ip_address is not None:
1224             mgmt_intf.ip_address = ip_address
1225         if port is not None:
1226             mgmt_intf.port = port
1227
1228         vnfr_dict = {"id": self._vnfr_id,
1229                      "nsr_id_ref": self._vnfr_msg.nsr_id_ref,
1230                      "name": self.name,
1231                      "member_vnf_index_ref": self.member_vnf_index,
1232                      "vnfd_ref": self.vnfd_id,
1233                      "operational_status": self.operational_status,
1234                      "operational_status_details": self._state_failed_reason,
1235                      "cloud_account": self.cloud_account_name,
1236                      "config_status": self._config_status
1237                      }
1238
1239         vnfr_dict.update(vnfd_copy_dict)
1240
1241         vnfr_msg = RwVnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr.from_dict(vnfr_dict)
1242         vnfr_msg.mgmt_interface = mgmt_intf
1243
1244         # Add all the VLRs  to  VNFR
1245         for vlr in self._vlrs:
1246             ivlr = vnfr_msg.internal_vlr.add()
1247             ivlr.vlr_ref = vlr.vlr_id
1248
1249         # Add all the VDURs to VDUR
1250         if self._vdus is not None:
1251             for vdu in self._vdus:
1252                 vdur = vnfr_msg.vdur.add()
1253                 vdur.from_dict(vdu.msg.as_dict())
1254
1255         if self.vnfd.msg.mgmt_interface.has_field('dashboard_params'):
1256             vnfr_msg.dashboard_url = self.dashboard_url
1257
1258         for cpr in self._cprs:
1259             new_cp = VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_ConnectionPoint.from_dict(cpr.as_dict())
1260             vnfr_msg.connection_point.append(new_cp)
1261
1262         if self._vnf_mon is not None:
1263             for monp in self._vnf_mon.msg:
1264                 vnfr_msg.monitoring_param.append(
1265                     VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_MonitoringParam.from_dict(monp.as_dict()))
1266
1267         if self._vnfr.vnf_configuration is not None:
1268             vnfr_msg.vnf_configuration.from_dict(self._vnfr.vnf_configuration.as_dict())
1269             if (ip_address is not None and
1270                     vnfr_msg.vnf_configuration.config_access.mgmt_ip_address is None):
1271                 vnfr_msg.vnf_configuration.config_access.mgmt_ip_address = ip_address
1272
1273         for group in self._vnfr_msg.placement_groups_info:
1274             group_info = VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_PlacementGroupsInfo()
1275             group_info.from_dict(group.as_dict())
1276             vnfr_msg.placement_groups_info.append(group_info)
1277
1278         return vnfr_msg
1279
1280     @property
1281     def dashboard_url(self):
1282         ip, cfg_port = self.mgmt_intf_info()
1283         protocol = 'http'
1284         http_port = 80
1285         if self.vnfd.msg.mgmt_interface.dashboard_params.has_field('https'):
1286             if self.vnfd.msg.mgmt_interface.dashboard_params.https is True:
1287                 protocol = 'https'
1288                 http_port = 443
1289         if self.vnfd.msg.mgmt_interface.dashboard_params.has_field('port'):
1290             http_port = self.vnfd.msg.mgmt_interface.dashboard_params.port
1291
1292         url = "{protocol}://{ip_address}:{port}/{path}".format(
1293                 protocol=protocol,
1294                 ip_address=ip,
1295                 port=http_port,
1296                 path=self.vnfd.msg.mgmt_interface.dashboard_params.path.lstrip("/"),
1297                 )
1298
1299         return url
1300
1301     @property
1302     def xpath(self):
1303         """ path for this  VNFR """
1304         return("D,/vnfr:vnfr-catalog"
1305                "/vnfr:vnfr[vnfr:id='{}']".format(self.vnfr_id))
1306
1307     @asyncio.coroutine
1308     def publish(self, xact):
1309         """ publish this VNFR """
1310         vnfr = self.msg
1311         self._log.debug("Publishing VNFR path = [%s], record = [%s]",
1312                         self.xpath, self.msg)
1313         vnfr.create_time = self._create_time
1314         yield from self._vnfm.publish_vnfr(xact, self.xpath, self.msg)
1315         self._log.debug("Published VNFR path = [%s], record = [%s]",
1316                         self.xpath, self.msg)
1317
1318     @asyncio.coroutine
1319     def create_vls(self):
1320         """ Publish The VLs associated with this VNF """
1321         self._log.debug("Publishing Internal Virtual Links for vnfd id: %s",
1322                         self.vnfd_id)
1323         for ivld_msg in self.vnfd.msg.internal_vld:
1324             self._log.debug("Creating internal vld:"
1325                             " %s, int_cp_ref = %s",
1326                             ivld_msg, ivld_msg.internal_connection_point
1327                             )
1328             vlr = InternalVirtualLinkRecord(dts=self._dts,
1329                                             log=self._log,
1330                                             loop=self._loop,
1331                                             ivld_msg=ivld_msg,
1332                                             vnfr_name=self.name,
1333                                             cloud_account_name=self.cloud_account_name
1334                                             )
1335             self._vlrs.append(vlr)
1336
1337             for int_cp in ivld_msg.internal_connection_point:
1338                 if int_cp.id_ref in self._vlr_by_cp:
1339                     msg = ("Connection point %s already "
1340                            " bound %s" % (int_cp.id_ref, self._vlr_by_cp[int_cp.id_ref]))
1341                     raise InternalVirtualLinkRecordError(msg)
1342                 self._log.debug("Setting vlr %s to internal cp = %s",
1343                                 vlr, int_cp.id_ref)
1344                 self._vlr_by_cp[int_cp.id_ref] = vlr
1345
1346     @asyncio.coroutine
1347     def instantiate_vls(self, xact, restart_mode=False):
1348         """ Instantiate the VLs associated with this VNF """
1349         self._log.debug("Instantiating Internal Virtual Links for vnfd id: %s",
1350                         self.vnfd_id)
1351
1352         for vlr in self._vlrs:
1353             self._log.debug("Instantiating VLR %s", vlr)
1354             yield from vlr.instantiate(xact, restart_mode)
1355
1356     def find_vlr_by_cp(self, cp_name):
1357         """ Find the VLR associated with the cp name """
1358         return self._vlr_by_cp[cp_name]
1359
1360     def resolve_placement_group_cloud_construct(self, input_group, nsr_config):
1361         """
1362         Returns the cloud specific construct for placement group
1363         Arguments:
1364             input_group: VNFD PlacementGroup
1365             nsr_config: Configuration for VNFDGroup MAP in the NSR config
1366         """
1367         copy_dict = ['name', 'requirement', 'strategy']
1368         for group_info in nsr_config.vnfd_placement_group_maps:
1369             if group_info.placement_group_ref == input_group.name and \
1370                group_info.vnfd_id_ref == self.vnfd_id:
1371                 group = VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_Vdur_PlacementGroupsInfo()
1372                 group_dict = {k:v for k,v in
1373                               group_info.as_dict().items()
1374                               if (k != 'placement_group_ref' and k !='vnfd_id_ref')}
1375                 for param in copy_dict:
1376                     group_dict.update({param: getattr(input_group, param)})
1377                 group.from_dict(group_dict)
1378                 return group
1379         return None
1380
1381     @asyncio.coroutine
1382     def get_vdu_placement_groups(self, vdu):
1383         placement_groups = []
1384         ### Step-1: Get VNF level placement groups
1385         for group in self._vnfr_msg.placement_groups_info:
1386             #group_info = VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_Vdur_PlacementGroupsInfo()
1387             #group_info.from_dict(group.as_dict())
1388             placement_groups.append(group)
1389
1390         ### Step-2: Get NSR config. This is required for resolving placement_groups cloud constructs
1391         nsr_config = yield from self.get_nsr_config()
1392
1393         ### Step-3: Get VDU level placement groups
1394         for group in self.vnfd.msg.placement_groups:
1395             for member_vdu in group.member_vdus:
1396                 if member_vdu.member_vdu_ref == vdu.id:
1397                     group_info = self.resolve_placement_group_cloud_construct(group,
1398                                                                               nsr_config)
1399                     if group_info is None:
1400                         self._log.info("Could not resolve cloud-construct for placement group: %s", group.name)
1401                         ### raise VNFMPlacementGroupError("Could not resolve cloud-construct for placement group: {}".format(group.name))
1402                     else:
1403                         self._log.info("Successfully resolved cloud construct for placement group: %s for VDU: %s in VNF: %s (Member Index: %s)",
1404                                        str(group_info),
1405                                        vdu.name,
1406                                        self.vnf_name,
1407                                        self.member_vnf_index)
1408                         placement_groups.append(group_info)
1409
1410         return placement_groups
1411
1412     @asyncio.coroutine
1413     def create_vdus(self, vnfr, restart_mode=False):
1414         """ Create the VDUs associated with this VNF """
1415
1416         def get_vdur_id(vdud):
1417             """Get the corresponding VDUR's id for the VDUD. This is useful in
1418             case of a restart.
1419
1420             In restart mode we check for exiting VDUR's ID and use them, if
1421             available. This way we don't end up creating duplicate VDURs
1422             """
1423             vdur_id = None
1424
1425             if restart_mode and vdud is not None:
1426                 try:
1427                     vdur = [vdur.id for vdur in vnfr._vnfr.vdur if vdur.vdu_id_ref == vdud.id]
1428                     vdur_id = vdur[0]
1429                 except IndexError:
1430                     self._log.error("Unable to find a VDUR for VDUD {}".format(vdud))
1431
1432             return vdur_id
1433
1434
1435         self._log.info("Creating VDU's for vnfd id: %s", self.vnfd_id)
1436         for vdu in self.vnfd.msg.vdu:
1437             self._log.debug("Creating vdu: %s", vdu)
1438             vdur_id = get_vdur_id(vdu)
1439
1440             placement_groups = yield from self.get_vdu_placement_groups(vdu)
1441             self._log.info("Launching VDU: %s from VNFD :%s (Member Index: %s) with Placement Groups: %s",
1442                            vdu.name,
1443                            self.vnf_name,
1444                            self.member_vnf_index,
1445                            [ group.name for group in placement_groups])
1446
1447             vdur = VirtualDeploymentUnitRecord(
1448                 dts=self._dts,
1449                 log=self._log,
1450                 loop=self._loop,
1451                 vdud=vdu,
1452                 vnfr=vnfr,
1453                 mgmt_intf=self.has_mgmt_interface(vdu),
1454                 cloud_account_name=self.cloud_account_name,
1455                 vnfd_package_store=self._vnfd_package_store,
1456                 vdur_id=vdur_id,
1457                 placement_groups = placement_groups,
1458                 )
1459             yield from vdur.vdu_opdata_register()
1460
1461             self._vdus.append(vdur)
1462
1463     @asyncio.coroutine
1464     def instantiate_vdus(self, xact, vnfr):
1465         """ Instantiate the VDUs associated with this VNF """
1466         self._log.debug("Instantiating VDU's for vnfd id %s: %s", self.vnfd_id, self._vdus)
1467
1468         lookup = {vdu.vdu_id: vdu for vdu in self._vdus}
1469
1470         # Identify any dependencies among the VDUs
1471         dependencies = collections.defaultdict(list)
1472         vdu_id_pattern = re.compile(r"\{\{ vdu\[([^]]+)\]\S* \}\}")
1473
1474         for vdu in self._vdus:
1475             if vdu.vdud_cloud_init is not None:
1476                 for vdu_id in vdu_id_pattern.findall(vdu.vdud_cloud_init):
1477                     if vdu_id != vdu.vdu_id:
1478                         # This means that vdu.vdu_id depends upon vdu_id,
1479                         # i.e. vdu_id must be instantiated before
1480                         # vdu.vdu_id.
1481                         dependencies[vdu.vdu_id].append(lookup[vdu_id])
1482
1483         # Define the terminal states of VDU instantiation
1484         terminal = (
1485                 VDURecordState.READY,
1486                 VDURecordState.TERMINATED,
1487                 VDURecordState.FAILED,
1488                 )
1489
1490         datastore = VdurDatastore()
1491         processed = set()
1492
1493         @asyncio.coroutine
1494         def instantiate_monitor(vdu):
1495             """Monitor the state of the VDU during instantiation
1496
1497             Arguments:
1498                 vdu - a VirtualDeploymentUnitRecord
1499
1500             """
1501             # wait for the VDUR to enter a terminal state
1502             while vdu._state not in terminal:
1503                 yield from asyncio.sleep(1, loop=self._loop)
1504
1505             # update the datastore
1506             datastore.update(vdu)
1507
1508             # add the VDU to the set of processed VDUs
1509             processed.add(vdu.vdu_id)
1510
1511         @asyncio.coroutine
1512         def instantiate(vdu):
1513             """Instantiate the specified VDU
1514
1515             Arguments:
1516                 vdu - a VirtualDeploymentUnitRecord
1517
1518             Raises:
1519                 if the VDU, or any of the VDUs this VDU depends upon, are
1520                 terminated or fail to instantiate properly, a
1521                 VirtualDeploymentUnitRecordError is raised.
1522
1523             """
1524             for dependency in dependencies[vdu.vdu_id]:
1525                 self._log.debug("{}: waiting for {}".format(vdu.vdu_id, dependency.vdu_id))
1526
1527                 while dependency.vdu_id not in processed:
1528                     yield from asyncio.sleep(1, loop=self._loop)
1529
1530                 if not dependency.active:
1531                     raise VirtualDeploymentUnitRecordError()
1532
1533             self._log.debug('instantiating {}'.format(vdu.vdu_id))
1534
1535             # Populate the datastore with the current values of the VDU
1536             datastore.add(vdu)
1537
1538             # Substitute any variables contained in the cloud config script
1539             config = str(vdu.vdud_cloud_init)
1540
1541             parts = re.split("\{\{ ([^\}]+) \}\}", config)
1542             if len(parts) > 1:
1543
1544                 # Extract the variable names
1545                 variables = list()
1546                 for variable in parts[1::2]:
1547                     variables.append(variable.lstrip('{{').rstrip('}}').strip())
1548
1549                 # Iterate of the variables and substitute values from the
1550                 # datastore.
1551                 for variable in variables:
1552
1553                     # Handle a reference to a VDU by ID
1554                     if variable.startswith('vdu['):
1555                         value = datastore.get(variable)
1556                         if value is None:
1557                             msg = "Unable to find a substitute for {} in {} cloud-init script"
1558                             raise ValueError(msg.format(variable, vdu.vdu_id))
1559
1560                         config = config.replace("{{ %s }}" % variable, value)
1561                         continue
1562
1563                     # Handle a reference to the current VDU
1564                     if variable.startswith('vdu'):
1565                         value = datastore.get('vdu[{}]'.format(vdu.vdu_id) + variable[3:])
1566                         config = config.replace("{{ %s }}" % variable, value)
1567                         continue
1568
1569                     # Handle unrecognized variables
1570                     msg = 'unrecognized cloud-config variable: {}'
1571                     raise ValueError(msg.format(variable))
1572
1573             # Instantiate the VDU
1574             with self._dts.transaction() as xact:
1575                 self._log.debug("Instantiating vdu: %s", vdu)
1576                 yield from vdu.instantiate(xact, vnfr, config=config)
1577                 if self._state == VirtualNetworkFunctionRecordState.FAILED:
1578                     self._log.error("Instatiation of VNF %s failed while instantiating vdu %s",
1579                                     self.vnfr_id, vdu)
1580
1581         # First create a set of tasks to monitor the state of the VDUs and
1582         # report when they have entered a terminal state
1583         for vdu in self._vdus:
1584             self._loop.create_task(instantiate_monitor(vdu))
1585
1586         for vdu in self._vdus:
1587             self._loop.create_task(instantiate(vdu))
1588
1589     def has_mgmt_interface(self, vdu):
1590         # ## TODO: Support additional mgmt_interface type options
1591         if self.vnfd.msg.mgmt_interface.vdu_id == vdu.id:
1592             return True
1593         return False
1594
1595     def vlr_xpath(self, vlr_id):
1596         """ vlr xpath """
1597         return(
1598             "D,/vlr:vlr-catalog/"
1599             "vlr:vlr[vlr:id = '{}']".format(vlr_id))
1600
1601     def ext_vlr_by_id(self, vlr_id):
1602         """ find ext vlr by id """
1603         return self._ext_vlrs[vlr_id]
1604
1605     @asyncio.coroutine
1606     def publish_inventory(self, xact):
1607         """ Publish the inventory associated with this VNF """
1608         self._log.debug("Publishing inventory for VNFR id: %s", self._vnfr_id)
1609
1610         for component in self.vnfd.msg.component:
1611             self._log.debug("Creating inventory component %s", component)
1612             mangled_name = VcsComponent.mangle_name(component.component_name,
1613                                                     self.vnf_name,
1614                                                     self.vnfd_id
1615                                                     )
1616             comp = VcsComponent(dts=self._dts,
1617                                 log=self._log,
1618                                 loop=self._loop,
1619                                 cluster_name=self._cluster_name,
1620                                 vcs_handler=self._vcs_handler,
1621                                 component=component,
1622                                 mangled_name=mangled_name,
1623                                 )
1624             if comp.name in self._inventory:
1625                 self._log.debug("Duplicate entries in inventory  %s for vnfr %s",
1626                                 component, self._vnfd_id)
1627                 return
1628             self._log.debug("Adding component %s for vnrf %s",
1629                             comp.name, self._vnfr_id)
1630             self._inventory[comp.name] = comp
1631             yield from comp.publish(xact)
1632
1633     def all_vdus_active(self):
1634         """ Are all VDUS in this VNFR active? """
1635         for vdu in self._vdus:
1636             if not vdu.active:
1637                 return False
1638
1639         self._log.debug("Inside all_vdus_active. Returning True")
1640         return True
1641
1642     @asyncio.coroutine
1643     def instantiation_failed(self, failed_reason=None):
1644         """ VNFR instantiation failed """
1645         self._log.debug("VNFR %s instantiation failed ", self.vnfr_id)
1646         self.set_state(VirtualNetworkFunctionRecordState.FAILED)
1647         self._state_failed_reason = failed_reason
1648
1649         # Update the VNFR with the changed status
1650         yield from self.publish(None)
1651
1652     @asyncio.coroutine
1653     def is_ready(self):
1654         """ This VNF is ready"""
1655         self._log.debug("VNFR id %s is ready", self.vnfr_id)
1656
1657         if self._state != VirtualNetworkFunctionRecordState.FAILED:
1658             self.set_state(VirtualNetworkFunctionRecordState.READY)
1659
1660         else:
1661             self._log.debug("VNFR id %s ignoring state change", self.vnfr_id)
1662
1663         # Update the VNFR with the changed status
1664         yield from self.publish(None)
1665
1666     def update_cp(self, cp_name, ip_address, mac_addr, cp_id):
1667         """Updated the connection point with ip address"""
1668         for cp in self._cprs:
1669             if cp.name == cp_name:
1670                 self._log.debug("Setting ip address and id for cp %s, cpr %s with ip %s id %s",
1671                                 cp_name, cp, ip_address, cp_id)
1672                 cp.ip_address = ip_address
1673                 cp.mac_address = mac_addr
1674                 cp.connection_point_id = cp_id
1675                 return
1676
1677         err = "No connection point %s found in VNFR id %s" % (cp.name, self._vnfr_id)
1678         self._log.debug(err)
1679         raise VirtualDeploymentUnitRecordError(err)
1680
1681     def set_state(self, state):
1682         """ Set state for this VNFR"""
1683         self._state = state
1684
1685     @asyncio.coroutine
1686     def instantiate(self, xact, restart_mode=False):
1687         """ instantiate this VNF """
1688         self.set_state(VirtualNetworkFunctionRecordState.VL_INIT_PHASE)
1689
1690         @asyncio.coroutine
1691         def fetch_vlrs():
1692             """ Fetch VLRs """
1693             # Iterate over all the connection points in VNFR and fetch the
1694             # associated VLRs
1695
1696             def cpr_from_cp(cp):
1697                 """ Creates a record level connection point from the desciptor cp"""
1698                 cp_fields = ["name", "image", "vm-flavor"]
1699                 cp_copy_dict = {k: v for k, v in cp.as_dict().items() if k in cp_fields}
1700                 cpr_dict = {}
1701                 cpr_dict.update(cp_copy_dict)
1702                 return VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_ConnectionPoint.from_dict(cpr_dict)
1703
1704             self._log.debug("Fetching VLRs for VNFR id = %s, cps = %s",
1705                             self._vnfr_id, self._vnfr.connection_point)
1706
1707             for cp in self._vnfr.connection_point:
1708                 cpr = cpr_from_cp(cp)
1709                 self._cprs.append(cpr)
1710                 self._log.debug("Adding Connection point record  %s ", cp)
1711
1712                 vlr_path = self.vlr_xpath(cp.vlr_ref)
1713                 self._log.debug("Fetching VLR with path = %s", vlr_path)
1714                 res_iter = yield from self._dts.query_read(self.vlr_xpath(cp.vlr_ref),
1715                                                            rwdts.XactFlag.MERGE)
1716                 for i in res_iter:
1717                     r = yield from i
1718                     d = r.result
1719                     self._ext_vlrs[cp.vlr_ref] = d
1720                     cpr.vlr_ref = cp.vlr_ref
1721                     self._log.debug("Fetched VLR [%s] with path = [%s]", d, vlr_path)
1722
1723         # Fetch the VNFD associated with the VNFR
1724         self._log.debug("VNFR-ID %s: Fetching vnfds", self._vnfr_id)
1725         self._vnfd = yield from self._vnfm.get_vnfd_ref(self._vnfd_id)
1726         self._log.debug("VNFR-ID %s: Fetched vnfd:%s", self._vnfr_id, self._vnfd)
1727
1728         assert self.vnfd is not None
1729
1730         # Fetch External VLRs
1731         self._log.debug("VNFR-ID %s: Fetching vlrs", self._vnfr_id)
1732         yield from fetch_vlrs()
1733
1734         # Publish inventory
1735         self._log.debug("VNFR-ID %s: Publishing Inventory", self._vnfr_id)
1736         yield from self.publish_inventory(xact)
1737
1738         # Publish inventory
1739         self._log.debug("VNFR-ID %s: Creating VLs", self._vnfr_id)
1740         yield from self.create_vls()
1741
1742         # publish the VNFR
1743         self._log.debug("VNFR-ID %s: Publish VNFR", self._vnfr_id)
1744         yield from self.publish(xact)
1745
1746         # instantiate VLs
1747         self._log.debug("VNFR-ID %s: Instantiate VLs", self._vnfr_id)
1748         try:
1749             yield from self.instantiate_vls(xact, restart_mode)
1750         except Exception as e:
1751             self._log.exception("VL instantiation failed (%s)", str(e))
1752             yield from self.instantiation_failed(str(e))
1753             return
1754
1755         self.set_state(VirtualNetworkFunctionRecordState.VM_INIT_PHASE)
1756
1757         # instantiate VDUs
1758         self._log.debug("VNFR-ID %s: Create VDUs", self._vnfr_id)
1759         yield from self.create_vdus(self, restart_mode)
1760
1761         # publish the VNFR
1762         self._log.debug("VNFR-ID %s: Publish VNFR", self._vnfr_id)
1763         yield from self.publish(xact)
1764
1765         # instantiate VDUs
1766         # ToDo: Check if this should be prevented during restart
1767         self._log.debug("VNFR-ID %s: Instantiate VDUs", self._vnfr_id)
1768         _ = self._loop.create_task(self.instantiate_vdus(xact, self))
1769
1770         # publish the VNFR
1771         self._log.debug("VNFR-ID %s: Publish VNFR", self._vnfr_id)
1772         yield from self.publish(xact)
1773
1774         self._log.debug("VNFR-ID %s: Instantiation Done", self._vnfr_id)
1775
1776     @asyncio.coroutine
1777     def terminate(self, xact):
1778         """ Terminate this virtual network function """
1779
1780         self._log.debug("Terminatng VNF id %s", self.vnfr_id)
1781
1782         self.set_state(VirtualNetworkFunctionRecordState.TERMINATE)
1783
1784         # stop monitoring
1785         if self._vnf_mon is not None:
1786             self._vnf_mon.stop()
1787             self._vnf_mon.deregister()
1788             self._vnf_mon = None
1789
1790         @asyncio.coroutine
1791         def terminate_vls():
1792             """ Terminate VLs in this VNF """
1793             for vl in self._vlrs:
1794                 yield from vl.terminate(xact)
1795
1796         @asyncio.coroutine
1797         def terminate_vdus():
1798             """ Terminate VDUS in this VNF """
1799             for vdu in self._vdus:
1800                 yield from vdu.terminate(xact)
1801
1802         self._log.debug("Terminatng VLs in VNF id %s", self.vnfr_id)
1803         self.set_state(VirtualNetworkFunctionRecordState.VL_TERMINATE_PHASE)
1804         yield from terminate_vls()
1805
1806         self._log.debug("Terminatng VDUs in VNF id %s", self.vnfr_id)
1807         self.set_state(VirtualNetworkFunctionRecordState.VDU_TERMINATE_PHASE)
1808         yield from terminate_vdus()
1809
1810         self._log.debug("Terminated  VNF id %s", self.vnfr_id)
1811         self.set_state(VirtualNetworkFunctionRecordState.TERMINATED)
1812
1813
1814 class VnfdDtsHandler(object):
1815     """ DTS handler for VNFD config changes """
1816     XPATH = "C,/vnfd:vnfd-catalog/vnfd:vnfd"
1817
1818     def __init__(self, dts, log, loop, vnfm):
1819         self._dts = dts
1820         self._log = log
1821         self._loop = loop
1822         self._vnfm = vnfm
1823         self._regh = None
1824
1825     @asyncio.coroutine
1826     def regh(self):
1827         """ DTS registration handle """
1828         return self._regh
1829
1830     @asyncio.coroutine
1831     def register(self):
1832         """ Register for VNFD configuration"""
1833
1834         def on_apply(dts, acg, xact, action, scratch):
1835             """Apply the  configuration"""
1836             self._log.debug("Got VNFM VNFD apply (xact: %s) (action: %s)(scr: %s)",
1837                             xact, action, scratch)
1838
1839             is_recovery = xact.xact is None and action == rwdts.AppconfAction.INSTALL
1840             # Create/Update a VNFD record
1841             for cfg in self._regh.get_xact_elements(xact):
1842                 # Only interested in those VNFD cfgs whose ID was received in prepare callback
1843                 if cfg.id in scratch.get('vnfds', []) or is_recovery:
1844                     self._vnfm.update_vnfd(cfg)
1845
1846             scratch.pop('vnfds', None)
1847
1848         @asyncio.coroutine
1849         def on_prepare(dts, acg, xact, xact_info, ks_path, msg, scratch):
1850             """ on prepare callback """
1851             self._log.debug("Got on prepare for VNFD (path: %s) (action: %s)",
1852                             ks_path.to_xpath(RwVnfmYang.get_schema()), msg)
1853             fref = ProtobufC.FieldReference.alloc()
1854             fref.goto_whole_message(msg.to_pbcm())
1855
1856             # Handle deletes in prepare_callback, but adds/updates in apply_callback
1857             if fref.is_field_deleted():
1858                 # Delete an VNFD record
1859                 self._log.debug("Deleting VNFD with id %s", msg.id)
1860                 if self._vnfm.vnfd_in_use(msg.id):
1861                     self._log.debug("Cannot delete VNFD in use - %s", msg)
1862                     err = "Cannot delete a VNFD in use - %s" % msg
1863                     raise VirtualNetworkFunctionDescriptorRefCountExists(err)
1864                 # Delete a VNFD record
1865                 yield from self._vnfm.delete_vnfd(msg.id)
1866             else:
1867                 # Handle actual adds/updates in apply_callback,
1868                 # just check if VNFD in use in prepare_callback
1869                 if self._vnfm.vnfd_in_use(msg.id):
1870                     self._log.debug("Cannot modify an VNFD in use - %s", msg)
1871                     err = "Cannot modify an VNFD in use - %s" % msg
1872                     raise VirtualNetworkFunctionDescriptorRefCountExists(err)
1873
1874                 # Add this VNFD to scratch to create/update in apply callback
1875                 vnfds = scratch.setdefault('vnfds', [])
1876                 vnfds.append(msg.id)
1877
1878             xact_info.respond_xpath(rwdts.XactRspCode.ACK)
1879
1880         self._log.debug(
1881             "Registering for VNFD config using xpath: %s",
1882             VnfdDtsHandler.XPATH,
1883             )
1884         acg_hdl = rift.tasklets.AppConfGroup.Handler(on_apply=on_apply)
1885         with self._dts.appconf_group_create(handler=acg_hdl) as acg:
1886             self._regh = acg.register(
1887                 xpath=VnfdDtsHandler.XPATH,
1888                 flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY,
1889                 on_prepare=on_prepare)
1890
1891
1892 class VcsComponentDtsHandler(object):
1893     """ Vcs Component DTS handler """
1894     XPATH = ("D,/rw-manifest:manifest" +
1895              "/rw-manifest:operational-inventory" +
1896              "/rw-manifest:component")
1897
1898     def __init__(self, dts, log, loop, vnfm):
1899         self._dts = dts
1900         self._log = log
1901         self._loop = loop
1902         self._regh = None
1903         self._vnfm = vnfm
1904
1905     @property
1906     def regh(self):
1907         """ DTS registration handle """
1908         return self._regh
1909
1910     @asyncio.coroutine
1911     def register(self):
1912         """ Registers VCS component dts publisher registration"""
1913         self._log.debug("VCS Comp publisher DTS handler registering path %s",
1914                         VcsComponentDtsHandler.XPATH)
1915
1916         hdl = rift.tasklets.DTS.RegistrationHandler()
1917         handlers = rift.tasklets.Group.Handler()
1918         with self._dts.group_create(handler=handlers) as group:
1919             self._regh = group.register(xpath=VcsComponentDtsHandler.XPATH,
1920                                         handler=hdl,
1921                                         flags=(rwdts.Flag.PUBLISHER |
1922                                                rwdts.Flag.NO_PREP_READ |
1923                                                rwdts.Flag.DATASTORE),)
1924
1925     @asyncio.coroutine
1926     def publish(self, xact, path, msg):
1927         """ Publishes the VCS component """
1928         self._log.debug("Publishing the VcsComponent xact = %s, %s:%s",
1929                         xact, path, msg)
1930         self.regh.create_element(path, msg)
1931         self._log.debug("Published the VcsComponent to %s xact = %s, %s:%s",
1932                         VcsComponentDtsHandler.XPATH, xact, path, msg)
1933
1934 class VnfrConsoleOperdataDtsHandler(object):
1935     """ registers 'D,/vnfr:vnfr-console/vnfr:vnfr[id]/vdur[id]' and handles CRUD from DTS"""
1936     @property
1937     def vnfr_vdu_console_xpath(self):
1938         """ path for resource-mgr"""
1939         return ("D,/rw-vnfr:vnfr-console/rw-vnfr:vnfr[rw-vnfr:id='{}']/rw-vnfr:vdur[vnfr:id='{}']".format(self._vnfr_id,self._vdur_id))
1940
1941     def __init__(self, dts, log, loop, vnfm, vnfr_id, vdur_id, vdu_id):
1942         self._dts = dts
1943         self._log = log
1944         self._loop = loop
1945         self._regh = None
1946         self._vnfm = vnfm
1947
1948         self._vnfr_id = vnfr_id
1949         self._vdur_id = vdur_id
1950         self._vdu_id = vdu_id
1951
1952     @asyncio.coroutine
1953     def register(self):
1954         """ Register for VNFR VDU Operational Data read from dts """
1955
1956         @asyncio.coroutine
1957         def on_prepare(xact_info, action, ks_path, msg):
1958             """ prepare callback from dts """
1959             xpath = ks_path.to_xpath(RwVnfrYang.get_schema())
1960             self._log.debug(
1961                 "Got VNFR VDU Opdata xact_info: %s, action: %s): %s:%s",
1962                 xact_info, action, xpath, msg
1963                 )
1964
1965             if action == rwdts.QueryAction.READ:
1966                 schema = RwVnfrYang.YangData_RwVnfr_VnfrConsole_Vnfr_Vdur.schema()
1967                 path_entry = schema.keyspec_to_entry(ks_path)
1968                 self._log.debug("VDU Opdata path is {}".format(path_entry))
1969                 try:
1970                     vnfr = self._vnfm.get_vnfr(self._vnfr_id)
1971                 except VnfRecordError as e:
1972                     self._log.error("VNFR id %s not found", self._vnfr_id)
1973                     xact_info.respond_xpath(rsp_code=rwdts.XactRspCode.ACK)
1974                     return
1975                 try:
1976                     vdur= vnfr._get_vdur_from_vdu_id(self._vdu_id)
1977                     if not vdur._state == VDURecordState.READY:
1978                         self._log.debug("VDUR state is not READY. current state is {}".format(vdur._state))
1979                         xact_info.respond_xpath(rsp_code=rwdts.XactRspCode.ACK)
1980                         return
1981                     with self._dts.transaction() as new_xact:
1982                         resp = yield from vdur.read_resource(new_xact)
1983                         vdur_console = RwVnfrYang.YangData_RwVnfr_VnfrConsole_Vnfr_Vdur()
1984                         vdur_console.id = self._vdur_id
1985                         if resp.console_url:
1986                             vdur_console.console_url = resp.console_url
1987                         else:
1988                             vdur_console.console_url = 'none'
1989                         self._log.debug("Recevied console URL for vdu {} is {}".format(self._vdu_id,vdur_console))
1990                 except Exception:
1991                     self._log.exception("Caught exception while reading VDU %s", self._vdu_id)
1992                     vdur_console = RwVnfrYang.YangData_RwVnfr_VnfrConsole_Vnfr_Vdur()
1993                     vdur_console.id = self._vdur_id
1994                     vdur_console.console_url = 'none'
1995
1996                 xact_info.respond_xpath(rsp_code=rwdts.XactRspCode.ACK,
1997                                             xpath=self.vnfr_vdu_console_xpath,
1998                                             msg=vdur_console)
1999             else:
2000                 #raise VnfRecordError("Not supported operation %s" % action)
2001                 self._log.error("Not supported operation %s" % action)
2002                 xact_info.respond_xpath(rsp_code=rwdts.XactRspCode.ACK)
2003                 return
2004
2005
2006         self._log.debug("Registering for VNFR VDU using xpath: %s",
2007                         self.vnfr_vdu_console_xpath)
2008         hdl = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_prepare,)
2009         with self._dts.group_create() as group:
2010             self._regh = group.register(xpath=self.vnfr_vdu_console_xpath,
2011                                         handler=hdl,
2012                                         flags=rwdts.Flag.PUBLISHER,
2013                                         )
2014
2015
2016 class VnfrDtsHandler(object):
2017     """ registers 'D,/vnfr:vnfr-catalog/vnfr:vnfr' and handles CRUD from DTS"""
2018     XPATH = "D,/vnfr:vnfr-catalog/vnfr:vnfr"
2019
2020     def __init__(self, dts, log, loop, vnfm):
2021         self._dts = dts
2022         self._log = log
2023         self._loop = loop
2024         self._vnfm = vnfm
2025
2026         self._regh = None
2027
2028     @property
2029     def regh(self):
2030         """ Return registration handle"""
2031         return self._regh
2032
2033     @property
2034     def vnfm(self):
2035         """ Return VNF manager instance """
2036         return self._vnfm
2037
2038     @asyncio.coroutine
2039     def register(self):
2040         """ Register for vnfr create/update/delete/read requests from dts """
2041         def on_commit(xact_info):
2042             """ The transaction has been committed """
2043             self._log.debug("Got vnfr commit (xact_info: %s)", xact_info)
2044             return rwdts.MemberRspCode.ACTION_OK
2045
2046         def on_abort(*args):
2047             """ Abort callback """
2048             self._log.debug("VNF  transaction got aborted")
2049
2050         @asyncio.coroutine
2051         def on_event(dts, g_reg, xact, xact_event, scratch_data):
2052
2053             @asyncio.coroutine
2054             def instantiate_realloc_vnfr(vnfr):
2055                 """Re-populate the vnfm after restart
2056
2057                 Arguments:
2058                     vlink
2059
2060                 """
2061
2062                 yield from vnfr.instantiate(None, restart_mode=True)
2063
2064             if xact_event == rwdts.MemberEvent.INSTALL:
2065                 curr_cfg = self.regh.elements
2066                 for cfg in curr_cfg:
2067                     vnfr = self.vnfm.create_vnfr(cfg)
2068                     self._loop.create_task(instantiate_realloc_vnfr(vnfr))
2069
2070             self._log.debug("Got on_event in vnfm")
2071
2072             return rwdts.MemberRspCode.ACTION_OK
2073
2074         @asyncio.coroutine
2075         def on_prepare(xact_info, action, ks_path, msg):
2076             """ prepare callback from dts """
2077             self._log.debug(
2078                 "Got vnfr on_prepare callback (xact_info: %s, action: %s): %s",
2079                 xact_info, action, msg
2080                 )
2081
2082             if action == rwdts.QueryAction.CREATE:
2083                 if not msg.has_field("vnfd_ref"):
2084                     err = "Vnfd reference not provided"
2085                     self._log.error(err)
2086                     raise VnfRecordError(err)
2087
2088                 vnfr = self.vnfm.create_vnfr(msg)
2089                 try:
2090                     # RIFT-9105: Unable to add a READ query under an existing transaction
2091                     # xact = xact_info.xact
2092                     yield from vnfr.instantiate(None)
2093                 except Exception as e:
2094                     self._log.exception(e)
2095                     self._log.error("Error while instantiating vnfr:%s", vnfr.vnfr_id)
2096                     vnfr.set_state(VirtualNetworkFunctionRecordState.FAILED)
2097                     yield from vnfr.publish(None)
2098             elif action == rwdts.QueryAction.DELETE:
2099                 schema = RwVnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr.schema()
2100                 path_entry = schema.keyspec_to_entry(ks_path)
2101                 vnfr = self._vnfm.get_vnfr(path_entry.key00.id)
2102
2103                 if vnfr is None:
2104                     self._log.debug("VNFR id %s not found for delete", path_entry.key00.id)
2105                     raise VirtualNetworkFunctionRecordNotFound(
2106                         "VNFR id %s", path_entry.key00.id)
2107
2108                 try:
2109                     yield from vnfr.terminate(xact_info.xact)
2110                     # Unref the VNFD
2111                     vnfr.vnfd.unref()
2112                     yield from self._vnfm.delete_vnfr(xact_info.xact, vnfr)
2113                 except Exception as e:
2114                     self._log.exception(e)
2115                     self._log.error("Caught exception while deleting vnfr %s", path_entry.key00.id)
2116
2117             elif action == rwdts.QueryAction.UPDATE:
2118                 schema = RwVnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr.schema()
2119                 path_entry = schema.keyspec_to_entry(ks_path)
2120                 vnfr = None
2121                 try:
2122                     vnfr = self._vnfm.get_vnfr(path_entry.key00.id)
2123                 except Exception as e:
2124                     self._log.debug("No vnfr found with id %s", path_entry.key00.id)
2125                     xact_info.respond_xpath(rwdts.XactRspCode.NA)
2126                     return
2127
2128                 if vnfr is None:
2129                     self._log.debug("VNFR id %s not found for update", path_entry.key00.id)
2130                     xact_info.respond_xpath(rwdts.XactRspCode.NA)
2131                     return
2132
2133                 self._log.debug("VNFR {} update config status {} (current {})".
2134                                 format(vnfr.name, msg.config_status, vnfr.config_status))
2135                 # Update the config status and publish
2136                 vnfr._config_status = msg.config_status
2137                 yield from vnfr.publish(None)
2138
2139             else:
2140                 raise NotImplementedError(
2141                     "%s action on VirtualNetworkFunctionRecord not supported",
2142                     action)
2143
2144             xact_info.respond_xpath(rwdts.XactRspCode.ACK)
2145
2146         self._log.debug("Registering for VNFR using xpath: %s",
2147                         VnfrDtsHandler.XPATH,)
2148
2149         hdl = rift.tasklets.DTS.RegistrationHandler(on_commit=on_commit,
2150                                                     on_prepare=on_prepare,)
2151         handlers = rift.tasklets.Group.Handler(on_event=on_event,)
2152         with self._dts.group_create(handler=handlers) as group:
2153             self._regh = group.register(xpath=VnfrDtsHandler.XPATH,
2154                                         handler=hdl,
2155                                         flags=(rwdts.Flag.PUBLISHER |
2156                                                rwdts.Flag.NO_PREP_READ |
2157                                                rwdts.Flag.CACHE |
2158                                                rwdts.Flag.DATASTORE),)
2159
2160     @asyncio.coroutine
2161     def create(self, xact, path, msg):
2162         """
2163         Create a VNFR record in DTS with path and message
2164         """
2165         self._log.debug("Creating VNFR xact = %s, %s:%s",
2166                         xact, path, msg)
2167
2168         self.regh.create_element(path, msg)
2169         self._log.debug("Created VNFR xact = %s, %s:%s",
2170                         xact, path, msg)
2171
2172     @asyncio.coroutine
2173     def update(self, xact, path, msg):
2174         """
2175         Update a VNFR record in DTS with path and message
2176         """
2177         self._log.debug("Updating VNFR xact = %s, %s:%s",
2178                         xact, path, msg)
2179         self.regh.update_element(path, msg)
2180         self._log.debug("Updated VNFR xact = %s, %s:%s",
2181                         xact, path, msg)
2182
2183     @asyncio.coroutine
2184     def delete(self, xact, path):
2185         """
2186         Delete a VNFR record in DTS with path and message
2187         """
2188         self._log.debug("Deleting VNFR xact = %s, %s", xact, path)
2189         self.regh.delete_element(path)
2190         self._log.debug("Deleted VNFR xact = %s, %s", xact, path)
2191
2192
2193 class VirtualNetworkFunctionDescriptor(object):
2194     """
2195     Virtual Network Function descriptor class
2196     """
2197
2198     def __init__(self, dts, log, loop, vnfm, vnfd):
2199         self._dts = dts
2200         self._log = log
2201         self._loop = loop
2202
2203         self._vnfm = vnfm
2204         self._vnfd = vnfd
2205         self._ref_count = 0
2206
2207     @property
2208     def ref_count(self):
2209         """ Returns the reference count associated with
2210             this Virtual Network Function Descriptor"""
2211         return self._ref_count
2212
2213     @property
2214     def id(self):
2215         """ Returns vnfd id """
2216         return self._vnfd.id
2217
2218     @property
2219     def name(self):
2220         """ Returns vnfd name """
2221         return self._vnfd.name
2222
2223     def in_use(self):
2224         """ Returns whether vnfd is in use or not """
2225         return True if self._ref_count > 0 else False
2226
2227     def ref(self):
2228         """ Take a reference on this object """
2229         self._ref_count += 1
2230         return self._ref_count
2231
2232     def unref(self):
2233         """ Release reference on this object """
2234         if self.ref_count < 1:
2235             msg = ("Unref on a VNFD object - vnfd id %s, ref_count = %s" %
2236                    (self.id, self._ref_count))
2237             self._log.critical(msg)
2238             raise VnfRecordError(msg)
2239         self._log.debug("Releasing ref on VNFD %s - curr ref_count:%s",
2240                         self.id, self.ref_count)
2241         self._ref_count -= 1
2242         return self._ref_count
2243
2244     @property
2245     def msg(self):
2246         """ Return the message associated with this NetworkServiceDescriptor"""
2247         return self._vnfd
2248
2249     @staticmethod
2250     def path_for_id(vnfd_id):
2251         """ Return path for the passed vnfd_id"""
2252         return "C,/vnfd:vnfd-catalog/vnfd:vnfd[vnfd:id = '{}']".format(vnfd_id)
2253
2254     def path(self):
2255         """ Return the path associated with this NetworkServiceDescriptor"""
2256         return VirtualNetworkFunctionDescriptor.path_for_id(self.id)
2257
2258     def update(self, vnfd):
2259         """ Update the Virtual Network Function Descriptor """
2260         if self.in_use():
2261             self._log.error("Cannot update descriptor %s in use refcnt=%d",
2262                             self.id, self.ref_count)
2263
2264             # The following loop is  added to debug RIFT-13284
2265             for vnf_rec in self._vnfm._vnfrs.values():
2266                 if vnf_rec.vnfd_id == self.id:
2267                     self._log.error("descriptor %s in used by %s:%s",
2268                                     self.id, vnf_rec.vnfr_id, vnf_rec.msg)
2269             raise VirtualNetworkFunctionDescriptorRefCountExists("Cannot update descriptor in use %s" % self.id)
2270         self._vnfd = vnfd
2271
2272     def delete(self):
2273         """ Delete the Virtual Network Function Descriptor """
2274         if self.in_use():
2275             self._log.error("Cannot delete descriptor %s in use refcnt=%d",
2276                             self.id)
2277
2278             # The following loop is  added to debug RIFT-13284
2279             for vnf_rec in self._vnfm._vnfrs.values():
2280                 if vnf_rec.vnfd_id == self.id:
2281                     self._log.error("descriptor %s in used by %s:%s",
2282                                     self.id, vnf_rec.vnfr_id, vnf_rec.msg)
2283             raise VirtualNetworkFunctionDescriptorRefCountExists("Cannot delete descriptor in use %s" % self.id)
2284         self._vnfm.delete_vnfd(self.id)
2285
2286
2287 class VnfdRefCountDtsHandler(object):
2288     """ The VNFD Ref Count DTS handler """
2289     XPATH = "D,/vnfr:vnfr-catalog/rw-vnfr:vnfd-ref-count"
2290
2291     def __init__(self, dts, log, loop, vnfm):
2292         self._dts = dts
2293         self._log = log
2294         self._loop = loop
2295         self._vnfm = vnfm
2296
2297         self._regh = None
2298
2299     @property
2300     def regh(self):
2301         """ Return registration handle """
2302         return self._regh
2303
2304     @property
2305     def vnfm(self):
2306         """ Return the NS manager instance """
2307         return self._vnfm
2308
2309     @asyncio.coroutine
2310     def register(self):
2311         """ Register for VNFD ref count read from dts """
2312
2313         @asyncio.coroutine
2314         def on_prepare(xact_info, action, ks_path, msg):
2315             """ prepare callback from dts """
2316             xpath = ks_path.to_xpath(RwVnfrYang.get_schema())
2317             self._log.debug(
2318                 "Got VNFD ref count get xact_info: %s, action: %s): %s:%s",
2319                 xact_info, action, xpath, msg
2320                 )
2321
2322             if action == rwdts.QueryAction.READ:
2323                 schema = RwVnfrYang.YangData_Vnfr_VnfrCatalog_VnfdRefCount.schema()
2324                 path_entry = schema.keyspec_to_entry(ks_path)
2325                 vnfd_list = yield from self._vnfm.get_vnfd_refcount(path_entry.key00.vnfd_id_ref)
2326                 for xpath, msg in vnfd_list:
2327                     self._log.debug("Responding to ref count query path:%s, msg:%s",
2328                                     xpath, msg)
2329                     xact_info.respond_xpath(rsp_code=rwdts.XactRspCode.MORE,
2330                                             xpath=xpath,
2331                                             msg=msg)
2332                 xact_info.respond_xpath(rwdts.XactRspCode.ACK)
2333             else:
2334                 raise VnfRecordError("Not supported operation %s" % action)
2335
2336         hdl = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_prepare,)
2337         with self._dts.group_create() as group:
2338             self._regh = group.register(xpath=VnfdRefCountDtsHandler.XPATH,
2339                                         handler=hdl,
2340                                         flags=rwdts.Flag.PUBLISHER,
2341                                         )
2342
2343
2344 class VdurDatastore(object):
2345     """
2346     This VdurDatastore is intended to expose select information about a VDUR
2347     such that it can be referenced in a cloud config file. The data that is
2348     exposed does not necessarily follow the structure of the data in the yang
2349     model. This is intentional. The data that are exposed are intended to be
2350     agnostic of the yang model so that changes in the model do not necessarily
2351     require changes to the interface provided to the user. It also means that
2352     the user does not need to be familiar with the RIFT.ware yang models.
2353     """
2354
2355     def __init__(self):
2356         """Create an instance of VdurDatastore"""
2357         self._vdur_data = dict()
2358         self._pattern = re.compile("vdu\[([^]]+)\]\.(.+)")
2359
2360     def add(self, vdur):
2361         """Add a new VDUR to the datastore
2362
2363         Arguments:
2364             vdur - a VirtualDeploymentUnitRecord instance
2365
2366         Raises:
2367             A ValueError is raised if the VDUR is (1) None or (2) already in
2368             the datastore.
2369
2370         """
2371         if vdur.vdu_id is None:
2372             raise ValueError('VDURs are required to have an ID')
2373
2374         if vdur.vdu_id in self._vdur_data:
2375             raise ValueError('cannot add a VDUR more than once')
2376
2377         self._vdur_data[vdur.vdu_id] = dict()
2378
2379         def set_if_not_none(key, attr):
2380             if attr is not None:
2381                 self._vdur_data[vdur.vdu_id][key] = attr
2382
2383         set_if_not_none('name', vdur._vdud.name)
2384         set_if_not_none('mgmt.ip', vdur.vm_management_ip)
2385
2386     def update(self, vdur):
2387         """Update the VDUR information in the datastore
2388
2389         Arguments:
2390             vdur - a GI representation of a VDUR
2391
2392         Raises:
2393             A ValueError is raised if the VDUR is (1) None or (2) already in
2394             the datastore.
2395
2396         """
2397         if vdur.vdu_id is None:
2398             raise ValueError('VNFDs are required to have an ID')
2399
2400         if vdur.vdu_id not in self._vdur_data:
2401             raise ValueError('VNF is not recognized')
2402
2403         def set_or_delete(key, attr):
2404             if attr is None:
2405                 if key in self._vdur_data[vdur.vdu_id]:
2406                     del self._vdur_data[vdur.vdu_id][key]
2407
2408             else:
2409                 self._vdur_data[vdur.vdu_id][key] = attr
2410
2411         set_or_delete('name', vdur._vdud.name)
2412         set_or_delete('mgmt.ip', vdur.vm_management_ip)
2413
2414     def remove(self, vdur_id):
2415         """Remove all of the data associated with specified VDUR
2416
2417         Arguments:
2418             vdur_id - the identifier of a VNFD in the datastore
2419
2420         Raises:
2421             A ValueError is raised if the VDUR is not contained in the
2422             datastore.
2423
2424         """
2425         if vdur_id not in self._vdur_data:
2426             raise ValueError('VNF is not recognized')
2427
2428         del self._vdur_data[vdur_id]
2429
2430     def get(self, expr):
2431         """Retrieve VDUR information from the datastore
2432
2433         An expression should be of the form,
2434
2435             vdu[<id>].<attr>
2436
2437         where <id> is the VDUR ID (an unquoted UUID), and <attr> is the name of
2438         the exposed attribute that the user wishes to retrieve.
2439
2440         If the requested data is not available, None is returned.
2441
2442         Arguments:
2443             expr - a string that specifies the data to return
2444
2445         Raises:
2446             A ValueError is raised if the provided expression cannot be parsed.
2447
2448         Returns:
2449             The requested data or None
2450
2451         """
2452         result = self._pattern.match(expr)
2453         if result is None:
2454             raise ValueError('data expression not recognized ({})'.format(expr))
2455
2456         vdur_id, key = result.groups()
2457
2458         if vdur_id not in self._vdur_data:
2459             return None
2460
2461         return self._vdur_data[vdur_id].get(key, None)
2462
2463
2464 class VnfManager(object):
2465     """ The virtual network function manager class """
2466     def __init__(self, dts, log, loop, cluster_name):
2467         self._dts = dts
2468         self._log = log
2469         self._loop = loop
2470         self._cluster_name = cluster_name
2471
2472         self._vcs_handler = VcsComponentDtsHandler(dts, log, loop, self)
2473         self._vnfr_handler = VnfrDtsHandler(dts, log, loop, self)
2474
2475         self._dts_handlers = [VnfdDtsHandler(dts, log, loop, self),
2476                               self._vnfr_handler,
2477                               self._vcs_handler,
2478                               VnfdRefCountDtsHandler(dts, log, loop, self)]
2479         self._vnfrs = {}
2480         self._vnfds = {}
2481
2482     @property
2483     def vnfr_handler(self):
2484         """ VNFR dts handler """
2485         return self._vnfr_handler
2486
2487     @property
2488     def vcs_handler(self):
2489         """ VCS dts handler """
2490         return self._vcs_handler
2491
2492     @asyncio.coroutine
2493     def register(self):
2494         """ Register all static DTS handlers """
2495         for hdl in self._dts_handlers:
2496             yield from hdl.register()
2497
2498     @asyncio.coroutine
2499     def run(self):
2500         """ Run this VNFM instance """
2501         self._log.debug("Run VNFManager - registering static DTS handlers""")
2502         yield from self.register()
2503
2504     def get_vnfr(self, vnfr_id):
2505         """ get VNFR by vnfr id """
2506
2507         if vnfr_id not in self._vnfrs:
2508             raise VnfRecordError("VNFR id %s not found", vnfr_id)
2509
2510         return self._vnfrs[vnfr_id]
2511
2512     def create_vnfr(self, vnfr):
2513         """ Create a VNFR instance """
2514         if vnfr.id in self._vnfrs:
2515             msg = "Vnfr id %s already exists" % vnfr.id
2516             self._log.error(msg)
2517             raise VnfRecordError(msg)
2518
2519         self._log.info("Create VirtualNetworkFunctionRecord %s from vnfd_id: %s",
2520                        vnfr.id,
2521                        vnfr.vnfd_ref)
2522
2523         self._vnfrs[vnfr.id] = VirtualNetworkFunctionRecord(
2524             self._dts, self._log, self._loop, self._cluster_name, self, self.vcs_handler, vnfr
2525             )
2526         return self._vnfrs[vnfr.id]
2527
2528     @asyncio.coroutine
2529     def delete_vnfr(self, xact, vnfr):
2530         """ Create a VNFR instance """
2531         if vnfr.vnfr_id in self._vnfrs:
2532             self._log.debug("Deleting VNFR id %s", vnfr.vnfr_id)
2533             yield from self._vnfr_handler.delete(xact, vnfr.xpath)
2534             del self._vnfrs[vnfr.vnfr_id]
2535
2536     @asyncio.coroutine
2537     def fetch_vnfd(self, vnfd_id):
2538         """ Fetch VNFDs based with the vnfd id"""
2539         vnfd_path = VirtualNetworkFunctionDescriptor.path_for_id(vnfd_id)
2540         self._log.debug("Fetch vnfd with path %s", vnfd_path)
2541         vnfd = None
2542
2543         res_iter = yield from self._dts.query_read(vnfd_path, rwdts.XactFlag.MERGE)
2544
2545         for ent in res_iter:
2546             res = yield from ent
2547             vnfd = res.result
2548
2549         if vnfd is None:
2550             err = "Failed to get  Vnfd %s" % vnfd_id
2551             self._log.error(err)
2552             raise VnfRecordError(err)
2553
2554         self._log.debug("Fetched vnfd for path %s, vnfd - %s", vnfd_path, vnfd)
2555
2556         return vnfd
2557
2558     @asyncio.coroutine
2559     def get_vnfd_ref(self, vnfd_id):
2560         """ Get Virtual Network Function descriptor for the passed vnfd_id"""
2561         vnfd = yield from self.get_vnfd(vnfd_id)
2562         vnfd.ref()
2563         return vnfd
2564
2565     @asyncio.coroutine
2566     def get_vnfd(self, vnfd_id):
2567         """ Get Virtual Network Function descriptor for the passed vnfd_id"""
2568         vnfd = None
2569         if vnfd_id not in self._vnfds:
2570             self._log.error("Cannot find VNFD id:%s", vnfd_id)
2571             vnfd = yield from self.fetch_vnfd(vnfd_id)
2572
2573             if vnfd is None:
2574                 self._log.error("Cannot find VNFD id:%s", vnfd_id)
2575                 raise VirtualNetworkFunctionDescriptorError("Cannot find VNFD id:%s", vnfd_id)
2576
2577             if vnfd.id != vnfd_id:
2578                 self._log.error("Bad Recovery state {} found for {}".format(vnfd.id, vnfd_id))
2579                 raise VirtualNetworkFunctionDescriptorError("Bad Recovery state {} found for {}".format(vnfd.id, vnfd_id))
2580
2581             if vnfd.id not in self._vnfds:
2582                 self.create_vnfd(vnfd)
2583
2584         return self._vnfds[vnfd_id]
2585
2586     def vnfd_in_use(self, vnfd_id):
2587         """ Is this VNFD in use """
2588         self._log.debug("Is this VNFD in use - msg:%s", vnfd_id)
2589         if vnfd_id in self._vnfds:
2590             return self._vnfds[vnfd_id].in_use()
2591         return False
2592
2593     @asyncio.coroutine
2594     def publish_vnfr(self, xact, path, msg):
2595         """ Publish a VNFR """
2596         self._log.debug("publish_vnfr called with path %s, msg %s",
2597                         path, msg)
2598         yield from self.vnfr_handler.update(xact, path, msg)
2599
2600     def create_vnfd(self, vnfd):
2601         """ Create a virtual network function descriptor """
2602         self._log.debug("Create virtual networkfunction descriptor - %s", vnfd)
2603         if vnfd.id in self._vnfds:
2604             self._log.error("Cannot create VNFD %s -VNFD id already exists", vnfd)
2605             raise VirtualNetworkFunctionDescriptorError("VNFD already exists-%s", vnfd.id)
2606
2607         self._vnfds[vnfd.id] = VirtualNetworkFunctionDescriptor(self._dts,
2608                                                                 self._log,
2609                                                                 self._loop,
2610                                                                 self,
2611                                                                 vnfd)
2612         return self._vnfds[vnfd.id]
2613
2614     def update_vnfd(self, vnfd):
2615         """ update the Virtual Network Function descriptor """
2616         self._log.debug("Update virtual network function descriptor - %s", vnfd)
2617
2618         if vnfd.id not in self._vnfds:
2619             self._log.debug("No VNFD found - creating VNFD id = %s", vnfd.id)
2620             self.create_vnfd(vnfd)
2621         else:
2622             self._log.debug("Updating VNFD id = %s, vnfd = %s", vnfd.id, vnfd)
2623             self._vnfds[vnfd.id].update(vnfd)
2624
2625     @asyncio.coroutine
2626     def delete_vnfd(self, vnfd_id):
2627         """ Delete the Virtual Network Function descriptor with the passed id """
2628         self._log.debug("Deleting the virtual network function descriptor - %s", vnfd_id)
2629         if vnfd_id not in self._vnfds:
2630             self._log.debug("Delete VNFD failed - cannot find vnfd-id %s", vnfd_id)
2631             raise VirtualNetworkFunctionDescriptorNotFound("Cannot find %s", vnfd_id)
2632
2633         if self._vnfds[vnfd_id].in_use():
2634             self._log.debug("Cannot delete VNFD id %s reference exists %s",
2635                             vnfd_id,
2636                             self._vnfds[vnfd_id].ref_count)
2637             raise VirtualNetworkFunctionDescriptorRefCountExists(
2638                 "Cannot delete :%s, ref_count:%s",
2639                 vnfd_id,
2640                 self._vnfds[vnfd_id].ref_count)
2641
2642         # Remove any files uploaded with VNFD and stored under $RIFT_ARTIFACTS/libs/<id>
2643         try:
2644             rift_artifacts_dir = os.environ['RIFT_ARTIFACTS']
2645             vnfd_dir = os.path.join(rift_artifacts_dir, 'launchpad/libs', vnfd_id)
2646             if os.path.exists(vnfd_dir):
2647                 shutil.rmtree(vnfd_dir, ignore_errors=True)
2648         except Exception as e:
2649             self._log.error("Exception in cleaning up VNFD {}: {}".
2650                             format(self._vnfds[vnfd_id].name, e))
2651             self._log.exception(e)
2652
2653         del self._vnfds[vnfd_id]
2654
2655     def vnfd_refcount_xpath(self, vnfd_id):
2656         """ xpath for ref count entry """
2657         return (VnfdRefCountDtsHandler.XPATH +
2658                 "[rw-vnfr:vnfd-id-ref = '{}']").format(vnfd_id)
2659
2660     @asyncio.coroutine
2661     def get_vnfd_refcount(self, vnfd_id):
2662         """ Get the vnfd_list from this VNFM"""
2663         vnfd_list = []
2664         if vnfd_id is None or vnfd_id == "":
2665             for vnfd in self._vnfds.values():
2666                 vnfd_msg = RwVnfrYang.YangData_Vnfr_VnfrCatalog_VnfdRefCount()
2667                 vnfd_msg.vnfd_id_ref = vnfd.id
2668                 vnfd_msg.instance_ref_count = vnfd.ref_count
2669                 vnfd_list.append((self.vnfd_refcount_xpath(vnfd.id), vnfd_msg))
2670         elif vnfd_id in self._vnfds:
2671                 vnfd_msg.vnfd_id_ref = self._vnfds[vnfd_id].id
2672                 vnfd_msg.instance_ref_count = self._vnfds[vnfd_id].ref_count
2673                 vnfd_list.append((self.vnfd_refcount_xpath(vnfd_id), vnfd_msg))
2674
2675         return vnfd_list
2676
2677
2678 class VnfmTasklet(rift.tasklets.Tasklet):
2679     """ VNF Manager tasklet class """
2680     def __init__(self, *args, **kwargs):
2681         super(VnfmTasklet, self).__init__(*args, **kwargs)
2682         self.rwlog.set_category("rw-mano-log")
2683         self.rwlog.set_subcategory("vnfm")
2684
2685         self._dts = None
2686         self._vnfm = None
2687
2688     def start(self):
2689         try:
2690             super(VnfmTasklet, self).start()
2691             self.log.info("Starting VnfmTasklet")
2692
2693             self.log.setLevel(logging.DEBUG)
2694
2695             self.log.debug("Registering with dts")
2696             self._dts = rift.tasklets.DTS(self.tasklet_info,
2697                                           RwVnfmYang.get_schema(),
2698                                           self.loop,
2699                                           self.on_dts_state_change)
2700
2701             self.log.debug("Created DTS Api GI Object: %s", self._dts)
2702         except Exception:
2703             print("Caught Exception in VNFM start:", sys.exc_info()[0])
2704             raise
2705
2706     def on_instance_started(self):
2707         """ Task insance started callback """
2708         self.log.debug("Got instance started callback")
2709
2710     def stop(self):
2711         try:
2712             self._dts.deinit()
2713         except Exception:
2714             print("Caught Exception in VNFM stop:", sys.exc_info()[0])
2715             raise
2716
2717     @asyncio.coroutine
2718     def init(self):
2719         """ Task init callback """
2720         try:
2721             vm_parent_name = self.tasklet_info.get_parent_vm_parent_instance_name()
2722             assert vm_parent_name is not None
2723             self._vnfm = VnfManager(self._dts, self.log, self.loop, vm_parent_name)
2724             yield from self._vnfm.run()
2725         except Exception:
2726             print("Caught Exception in VNFM init:", sys.exc_info()[0])
2727             raise
2728
2729     @asyncio.coroutine
2730     def run(self):
2731         """ Task run callback """
2732         pass
2733
2734     @asyncio.coroutine
2735     def on_dts_state_change(self, state):
2736         """Take action according to current dts state to transition
2737         application into the corresponding application state
2738
2739         Arguments
2740             state - current dts state
2741         """
2742         switch = {
2743             rwdts.State.INIT: rwdts.State.REGN_COMPLETE,
2744             rwdts.State.CONFIG: rwdts.State.RUN,
2745         }
2746
2747         handlers = {
2748             rwdts.State.INIT: self.init,
2749             rwdts.State.RUN: self.run,
2750         }
2751
2752         # Transition application to next state
2753         handler = handlers.get(state, None)
2754         if handler is not None:
2755             yield from handler()
2756
2757         # Transition dts to next state
2758         next_state = switch.get(state, None)
2759         if next_state is not None:
2760             self._dts.handle.set_state(next_state)