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