2 # Copyright 2016 RIFT.IO Inc
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
8 # http://www.apache.org/licenses/LICENSE-2.0
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.
17 # vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
20 import ncclient
.asyncio_manager
32 from collections
import deque
33 from collections
import defaultdict
37 gi
.require_version('RwYang', '1.0')
38 gi
.require_version('ProjectNsdYang', '1.0')
39 gi
.require_version('RwDts', '1.0')
40 gi
.require_version('RwNsmYang', '1.0')
41 gi
.require_version('RwNsrYang', '1.0')
42 gi
.require_version('RwTypes', '1.0')
43 gi
.require_version('RwVlrYang', '1.0')
44 gi
.require_version('RwVnfrYang', '1.0')
45 from gi
.repository
import (
49 ProjectNsdYang
as NsdYang
,
61 import rift
.mano
.ncclient
62 import rift
.mano
.config_data
.config
63 import rift
.mano
.dts
as mano_dts
64 from rift
.mano
.utils
.project
import (
67 get_add_delete_update_cfgs
,
71 from . import rwnsm_conman
as conman
73 from . import publisher
75 from . import config_value_pool
76 from . import rwvnffgmgr
77 from . import scale_group
80 class NetworkServiceRecordState(Enum
):
81 """ Network Service Record State """
85 VNFFG_INIT_PHASE
= 104
91 VL_TERMINATE_PHASE
= 111
92 VNF_TERMINATE_PHASE
= 112
93 VNFFG_TERMINATE_PHASE
= 113
100 class NetworkServiceRecordError(Exception):
101 """ Network Service Record Error """
105 class NetworkServiceDescriptorError(Exception):
106 """ Network Service Descriptor Error """
110 class VirtualNetworkFunctionRecordError(Exception):
111 """ Virtual Network Function Record Error """
115 class NetworkServiceDescriptorNotFound(Exception):
116 """ Cannot find Network Service Descriptor"""
120 class NetworkServiceDescriptorRefCountExists(Exception):
121 """ Network Service Descriptor reference count exists """
125 class NetworkServiceDescriptorUnrefError(Exception):
126 """ Failed to unref a network service descriptor """
130 class NsrInstantiationFailed(Exception):
131 """ Failed to instantiate network service """
135 class VnfInstantiationFailed(Exception):
136 """ Failed to instantiate virtual network function"""
140 class VnffgInstantiationFailed(Exception):
141 """ Failed to instantiate virtual network function"""
145 class VnfDescriptorError(Exception):
146 """Failed to instantiate virtual network function"""
150 class ScalingOperationError(Exception):
154 class ScaleGroupMissingError(Exception):
158 class PlacementGroupError(Exception):
162 class NsrNsdUpdateError(Exception):
166 class NsrVlUpdateError(NsrNsdUpdateError
):
170 class VlRecordState(Enum
):
171 """ VL Record State """
173 INSTANTIATION_PENDING
= 102
175 TERMINATE_PENDING
= 104
180 class VnffgRecordState(Enum
):
181 """ VNFFG Record State """
183 INSTANTIATION_PENDING
= 102
185 TERMINATE_PENDING
= 104
190 class VnffgRecord(object):
191 """ Vnffg Records class"""
194 def __init__(self
, dts
, log
, loop
, vnffgmgr
, nsr
, nsr_name
, vnffgd_msg
, sdn_account_name
):
199 self
._vnffgmgr
= vnffgmgr
201 self
._nsr
_name
= nsr_name
202 self
._vnffgd
_msg
= vnffgd_msg
203 if sdn_account_name
is None:
204 self
._sdn
_account
_name
= ''
206 self
._sdn
_account
_name
= sdn_account_name
208 self
._vnffgr
_id
= str(uuid
.uuid4())
209 self
._vnffgr
_rsp
_id
= list()
210 self
._vnffgr
_state
= VnffgRecordState
.INIT
215 return self
._vnffgr
_id
219 """ state of this VNF """
220 return self
._vnffgr
_state
222 def fetch_vnffgr(self
):
224 Get VNFFGR message to be published
227 if self
._vnffgr
_state
== VnffgRecordState
.INIT
:
228 vnffgr_dict
= {"id": self
._vnffgr
_id
,
229 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
230 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
231 "sdn_account": self
._sdn
_account
_name
,
232 "operational_status": 'init',
234 vnffgr
= NsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
235 elif self
._vnffgr
_state
== VnffgRecordState
.TERMINATED
:
236 vnffgr_dict
= {"id": self
._vnffgr
_id
,
237 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
238 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
239 "sdn_account": self
._sdn
_account
_name
,
240 "operational_status": 'terminated',
242 vnffgr
= NsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
245 vnffgr
= self
._vnffgmgr
.fetch_vnffgr(self
._vnffgr
_id
)
247 self
._log
.exception("Fetching VNFFGR for VNFFG with id %s failed", self
._vnffgr
_id
)
248 self
._vnffgr
_state
= VnffgRecordState
.FAILED
249 vnffgr_dict
= {"id": self
._vnffgr
_id
,
250 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
251 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
252 "sdn_account": self
._sdn
_account
_name
,
253 "operational_status": 'failed',
255 vnffgr
= NsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
260 def vnffgr_create_msg(self
):
261 """ Virtual Link Record message for Creating VLR in VNS """
262 vnffgr_dict
= {"id": self
._vnffgr
_id
,
263 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
264 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
265 "sdn_account": self
._sdn
_account
_name
,
267 vnffgr
= NsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
268 for rsp
in self
._vnffgd
_msg
.rsp
:
269 vnffgr_rsp
= vnffgr
.rsp
.add()
270 vnffgr_rsp
.id = str(uuid
.uuid4())
271 vnffgr_rsp
.name
= self
._nsr
.name
+ '.' + rsp
.name
272 self
._vnffgr
_rsp
_id
.append(vnffgr_rsp
.id)
273 vnffgr_rsp
.vnffgd_rsp_id_ref
= rsp
.id
274 vnffgr_rsp
.vnffgd_rsp_name_ref
= rsp
.name
275 for rsp_cp_ref
in rsp
.vnfd_connection_point_ref
:
276 vnfd
= [vnfr
.vnfd
for vnfr
in self
._nsr
.vnfrs
.values() if vnfr
.vnfd
.id == rsp_cp_ref
.vnfd_id_ref
]
277 self
._log
.debug("VNFD message during VNFFG instantiation is %s",vnfd
)
278 if len(vnfd
) > 0 and vnfd
[0].has_field('service_function_type'):
279 self
._log
.debug("Service Function Type for VNFD ID %s is %s",
280 rsp_cp_ref
.vnfd_id_ref
, vnfd
[0].service_function_type
)
282 self
._log
.error("Service Function Type not available for VNFD ID %s; Skipping in chain",
283 rsp_cp_ref
.vnfd_id_ref
)
286 vnfr_cp_ref
= vnffgr_rsp
.vnfr_connection_point_ref
.add()
287 vnfr_cp_ref
.member_vnf_index_ref
= rsp_cp_ref
.member_vnf_index_ref
288 vnfr_cp_ref
.hop_number
= rsp_cp_ref
.order
289 vnfr_cp_ref
.vnfd_id_ref
=rsp_cp_ref
.vnfd_id_ref
290 vnfr_cp_ref
.service_function_type
= vnfd
[0].service_function_type
291 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
292 if (nsr_vnfr
.vnfd
.id == vnfr_cp_ref
.vnfd_id_ref
and
293 nsr_vnfr
.member_vnf_index
== vnfr_cp_ref
.member_vnf_index_ref
):
294 vnfr_cp_ref
.vnfr_id_ref
= nsr_vnfr
.id
295 vnfr_cp_ref
.vnfr_name_ref
= nsr_vnfr
.name
296 vnfr_cp_ref
.vnfr_connection_point_ref
= rsp_cp_ref
.vnfd_connection_point_ref
298 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
299 self
._log
.debug(" Received VNFR is %s", vnfr
)
300 while vnfr
.operational_status
!= 'running':
301 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
302 if vnfr
.operational_status
== 'failed':
303 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
304 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" %
306 yield from asyncio
.sleep(2, loop
=self
._loop
)
307 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
308 self
._log
.debug("Received VNFR is %s", vnfr
)
310 vnfr_cp_ref
.connection_point_params
.mgmt_address
= vnfr
.mgmt_interface
.ip_address
311 for cp
in vnfr
.connection_point
:
312 if cp
.name
== vnfr_cp_ref
.vnfr_connection_point_ref
:
313 vnfr_cp_ref
.connection_point_params
.port_id
= cp
.connection_point_id
314 vnfr_cp_ref
.connection_point_params
.name
= self
._nsr
.name
+ '.' + cp
.name
315 for vdu
in vnfr
.vdur
:
316 for ext_intf
in vdu
.external_interface
:
317 if ext_intf
.name
== vnfr_cp_ref
.vnfr_connection_point_ref
:
318 vnfr_cp_ref
.connection_point_params
.vm_id
= vdu
.vim_id
319 self
._log
.debug("VIM ID for CP %s in VNFR %s is %s",cp
.name
,nsr_vnfr
.id,
320 vnfr_cp_ref
.connection_point_params
.vm_id
)
323 vnfr_cp_ref
.connection_point_params
.address
= cp
.ip_address
324 vnfr_cp_ref
.connection_point_params
.port
= VnffgRecord
.SFF_DP_PORT
326 for vnffgd_classifier
in self
._vnffgd
_msg
.classifier
:
327 _rsp
= [rsp
for rsp
in vnffgr
.rsp
if rsp
.vnffgd_rsp_id_ref
== vnffgd_classifier
.rsp_id_ref
]
329 rsp_id_ref
= _rsp
[0].id
330 rsp_name
= _rsp
[0].name
332 self
._log
.error("RSP with ID %s not found during classifier creation for classifier id %s",
333 vnffgd_classifier
.rsp_id_ref
,vnffgd_classifier
.id)
335 vnffgr_classifier
= vnffgr
.classifier
.add()
336 vnffgr_classifier
.id = vnffgd_classifier
.id
337 vnffgr_classifier
.name
= self
._nsr
.name
+ '.' + vnffgd_classifier
.name
338 _rsp
[0].classifier_name
= vnffgr_classifier
.name
339 vnffgr_classifier
.rsp_id_ref
= rsp_id_ref
340 vnffgr_classifier
.rsp_name
= rsp_name
341 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
342 if (nsr_vnfr
.vnfd
.id == vnffgd_classifier
.vnfd_id_ref
and
343 nsr_vnfr
.member_vnf_index
== vnffgd_classifier
.member_vnf_index_ref
):
344 vnffgr_classifier
.vnfr_id_ref
= nsr_vnfr
.id
345 vnffgr_classifier
.vnfr_name_ref
= nsr_vnfr
.name
346 vnffgr_classifier
.vnfr_connection_point_ref
= vnffgd_classifier
.vnfd_connection_point_ref
348 if nsr_vnfr
.vnfd
.service_function_chain
== 'CLASSIFIER':
349 vnffgr_classifier
.sff_name
= nsr_vnfr
.name
351 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
352 self
._log
.debug(" Received VNFR is %s", vnfr
)
353 while vnfr
.operational_status
!= 'running':
354 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
355 if vnfr
.operational_status
== 'failed':
356 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
357 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" %
359 yield from asyncio
.sleep(2, loop
=self
._loop
)
360 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
361 self
._log
.debug("Received VNFR is %s", vnfr
)
363 for cp
in vnfr
.connection_point
:
364 if cp
.name
== vnffgr_classifier
.vnfr_connection_point_ref
:
365 vnffgr_classifier
.port_id
= cp
.connection_point_id
366 vnffgr_classifier
.ip_address
= cp
.ip_address
367 for vdu
in vnfr
.vdur
:
368 for ext_intf
in vdu
.external_interface
:
369 if ext_intf
.name
== vnffgr_classifier
.vnfr_connection_point_ref
:
370 vnffgr_classifier
.vm_id
= vdu
.vim_id
371 self
._log
.debug("VIM ID for CP %s in VNFR %s is %s",
373 vnfr_cp_ref
.connection_point_params
.vm_id
)
376 self
._log
.info("VNFFGR msg to be sent is %s", vnffgr
)
380 def vnffgr_nsr_sff_list(self
):
381 """ SFF List for VNFR """
383 sf_list
= [nsr_vnfr
.name
for nsr_vnfr
in self
._nsr
.vnfrs
.values() if nsr_vnfr
.vnfd
.service_function_chain
== 'SF']
385 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
386 if (nsr_vnfr
.vnfd
.service_function_chain
== 'CLASSIFIER' or nsr_vnfr
.vnfd
.service_function_chain
== 'SFF'):
387 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
388 self
._log
.debug(" Received VNFR is %s", vnfr
)
389 while vnfr
.operational_status
!= 'running':
390 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
391 if vnfr
.operational_status
== 'failed':
392 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
393 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self
.id, vnfr
.id))
394 yield from asyncio
.sleep(2, loop
=self
._loop
)
395 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
396 self
._log
.debug("Received VNFR is %s", vnfr
)
398 sff
= RwsdnYang
.VNFFGSff()
399 sff_list
[nsr_vnfr
.vnfd
.id] = sff
400 sff
.name
= nsr_vnfr
.name
401 sff
.function_type
= nsr_vnfr
.vnfd
.service_function_chain
403 sff
.mgmt_address
= vnfr
.mgmt_interface
.ip_address
404 sff
.mgmt_port
= VnffgRecord
.SFF_MGMT_PORT
405 for cp
in vnfr
.connection_point
:
406 sff_dp
= sff
.dp_endpoints
.add()
407 sff_dp
.name
= self
._nsr
.name
+ '.' + cp
.name
408 sff_dp
.address
= cp
.ip_address
409 sff_dp
.port
= VnffgRecord
.SFF_DP_PORT
410 if nsr_vnfr
.vnfd
.service_function_chain
== 'SFF':
411 for sf_name
in sf_list
:
412 _sf
= sff
.vnfr_list
.add()
413 _sf
.vnfr_name
= sf_name
418 def instantiate(self
):
419 """ Instantiate this VNFFG """
421 self
._log
.info("Instaniating VNFFGR with vnffgd %s",
425 vnffgr_request
= yield from self
.vnffgr_create_msg()
426 vnffg_sff_list
= yield from self
.vnffgr_nsr_sff_list()
429 vnffgr
= self
._vnffgmgr
.create_vnffgr(vnffgr_request
,self
._vnffgd
_msg
.classifier
,vnffg_sff_list
)
430 except Exception as e
:
431 self
._log
.exception("VNFFG instantiation failed: %s", str(e
))
432 self
._vnffgr
_state
= VnffgRecordState
.FAILED
433 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFFGR %s failure" % (self
.id, vnffgr_request
.id))
435 self
._vnffgr
_state
= VnffgRecordState
.INSTANTIATION_PENDING
437 self
._log
.info("Instantiated VNFFGR :%s", vnffgr
)
438 self
._vnffgr
_state
= VnffgRecordState
.ACTIVE
440 self
._log
.info("Invoking update_state to update NSR state for NSR ID: %s", self
._nsr
.id)
441 yield from self
._nsr
.update_state()
443 def vnffgr_in_vnffgrm(self
):
444 """ Is there a VNFR record in VNFM """
445 if (self
._vnffgr
_state
== VnffgRecordState
.ACTIVE
or
446 self
._vnffgr
_state
== VnffgRecordState
.INSTANTIATION_PENDING
or
447 self
._vnffgr
_state
== VnffgRecordState
.FAILED
):
454 """ Terminate this VNFFGR """
455 if not self
.vnffgr_in_vnffgrm():
456 self
._log
.error("Ignoring terminate request for id %s in state %s",
457 self
.id, self
._vnffgr
_state
)
460 self
._log
.info("Terminating VNFFGR id:%s", self
.id)
461 self
._vnffgr
_state
= VnffgRecordState
.TERMINATE_PENDING
463 self
._vnffgmgr
.terminate_vnffgr(self
._vnffgr
_id
)
465 self
._vnffgr
_state
= VnffgRecordState
.TERMINATED
466 self
._log
.debug("Terminated VNFFGR id:%s", self
.id)
469 class VirtualLinkRecord(object):
470 """ Virtual Link Records class"""
471 XPATH
= "D,/vlr:vlr-catalog/vlr:vlr"
474 def create_record(dts
, log
, loop
, project
, nsr_name
, vld_msg
,
475 cloud_account_name
, om_datacenter
, ip_profile
,
476 nsr_id
, restart_mode
=False):
477 """Creates a new VLR object based on the given data.
479 If restart mode is enabled, then we look for existing records in the
480 DTS and create a VLR records using the exiting data(ID)
485 vlr_obj
= VirtualLinkRecord(
499 res_iter
= yield from dts
.query_read(
500 project
.add_project("D,/vlr:vlr-catalog/vlr:vlr"),
501 rwdts
.XactFlag
.MERGE
)
504 response
= yield from fut
505 vlr
= response
.result
507 # Check if the record is already present, if so use the ID of
508 # the existing record. Since the name of the record is uniquely
509 # formed we can use it as a search key!
510 if vlr
.name
== vlr_obj
.name
:
511 vlr_obj
.reset_id(vlr
.id)
516 def __init__(self
, dts
, log
, loop
, project
, nsr_name
, vld_msg
,
517 cloud_account_name
, om_datacenter
, ip_profile
, nsr_id
):
521 self
._project
= project
522 self
._nsr
_name
= nsr_name
523 self
._vld
_msg
= vld_msg
524 self
._cloud
_account
_name
= cloud_account_name
525 self
._om
_datacenter
_name
= om_datacenter
526 self
._assigned
_subnet
= None
527 self
._nsr
_id
= nsr_id
528 self
._ip
_profile
= ip_profile
529 self
._vlr
_id
= str(uuid
.uuid4())
530 self
._state
= VlRecordState
.INIT
531 self
._prev
_state
= None
532 self
._create
_time
= int(time
.time())
536 """ path for this object """
537 return self
._project
.add_project("D,/vlr:vlr-catalog/vlr:vlr[vlr:id = '{}']".
538 format(self
._vlr
_id
))
547 """ Get NSR name for this VL """
552 """ Virtual Link Desciptor """
556 def assigned_subnet(self
):
557 """ Subnet assigned to this VL"""
558 return self
._assigned
_subnet
563 Get the name for this VLR.
564 VLR name is "nsr name:VLD name"
566 if self
.vld_msg
.vim_network_name
:
567 return self
.vld_msg
.vim_network_name
568 elif self
.vld_msg
.name
== "multisite":
569 # This is a temporary hack to identify manually provisioned inter-site network
570 return self
.vld_msg
.name
572 return self
._nsr
_name
+ "." + self
.vld_msg
.name
575 def cloud_account_name(self
):
576 """ Cloud account that this VLR should be created in """
577 return self
._cloud
_account
_name
580 def om_datacenter_name(self
):
581 """ Datacenter that this VLR should be created in """
582 return self
._om
_datacenter
_name
586 """ Get the VLR path from VLR """
587 return (VirtualLinkRecord
.XPATH
+ "[vlr:id = '{}']").format(vlr
.id)
595 def state(self
, value
):
596 """ VLR set state """
600 def prev_state(self
):
601 """ VLR previous state """
602 return self
._prev
_state
605 def prev_state(self
, value
):
606 """ VLR set previous state """
607 self
._prev
_state
= value
611 """ Virtual Link Record message for Creating VLR in VNS """
612 vld_fields
= ["short_name",
620 vld_copy_dict
= {k
: v
for k
, v
in self
.vld_msg
.as_dict().items()
623 vlr_dict
= {"id": self
._vlr
_id
,
624 "nsr_id_ref": self
._nsr
_id
,
625 "vld_ref": self
.vld_msg
.id,
627 "create_time": self
._create
_time
,
628 "cloud_account": self
.cloud_account_name
,
629 "om_datacenter": self
.om_datacenter_name
,
632 if self
._ip
_profile
and self
._ip
_profile
.has_field('ip_profile_params'):
633 vlr_dict
['ip_profile_params' ] = self
._ip
_profile
.ip_profile_params
.as_dict()
635 vlr_dict
.update(vld_copy_dict
)
636 vlr
= RwVlrYang
.YangData_RwProject_Project_VlrCatalog_Vlr
.from_dict(vlr_dict
)
639 def reset_id(self
, vlr_id
):
640 self
._vlr
_id
= vlr_id
642 def create_nsr_vlr_msg(self
, vnfrs
):
643 """ The VLR message"""
644 nsr_vlr
= RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr_Vlr()
645 nsr_vlr
.vlr_ref
= self
._vlr
_id
646 nsr_vlr
.assigned_subnet
= self
.assigned_subnet
647 nsr_vlr
.cloud_account
= self
.cloud_account_name
648 nsr_vlr
.om_datacenter
= self
.om_datacenter_name
650 for conn
in self
.vld_msg
.vnfd_connection_point_ref
:
652 if (vnfr
.vnfd
.id == conn
.vnfd_id_ref
and
653 vnfr
.member_vnf_index
== conn
.member_vnf_index_ref
and
654 self
.cloud_account_name
== vnfr
.cloud_account_name
and
655 self
.om_datacenter_name
== vnfr
.om_datacenter_name
):
656 cp_entry
= nsr_vlr
.vnfr_connection_point_ref
.add()
657 cp_entry
.vnfr_id
= vnfr
.id
658 cp_entry
.connection_point
= conn
.vnfd_connection_point_ref
663 def instantiate(self
):
664 """ Instantiate this VL """
665 self
._log
.debug("Instaniating VLR key %s, vld %s",
666 self
.xpath
, self
._vld
_msg
)
668 self
._state
= VlRecordState
.INSTANTIATION_PENDING
669 self
._log
.debug("Executing VL create path:%s msg:%s",
670 self
.xpath
, self
.vlr_msg
)
672 with self
._dts
.transaction(flags
=0) as xact
:
673 block
= xact
.block_create()
674 block
.add_query_create(self
.xpath
, self
.vlr_msg
)
675 self
._log
.debug("Executing VL create path:%s msg:%s",
676 self
.xpath
, self
.vlr_msg
)
677 res_iter
= yield from block
.execute(now
=True)
683 self
._state
= VlRecordState
.FAILED
684 raise NsrInstantiationFailed("Failed NS %s instantiation due to empty response" % self
.id)
686 if vlr
.operational_status
== 'failed':
687 self
._log
.debug("NS Id:%s VL creation failed for vlr id %s", self
.id, vlr
.id)
688 self
._state
= VlRecordState
.FAILED
689 raise NsrInstantiationFailed("Failed VL %s instantiation (%s)" % (vlr
.id, vlr
.operational_status_details
))
691 self
._log
.info("Instantiated VL with xpath %s and vlr:%s",
693 self
._state
= VlRecordState
.ACTIVE
694 self
._assigned
_subnet
= vlr
.assigned_subnet
696 def vlr_in_vns(self
):
697 """ Is there a VLR record in VNS """
698 if (self
._state
== VlRecordState
.ACTIVE
or
699 self
._state
== VlRecordState
.INSTANTIATION_PENDING
or
700 self
._state
== VlRecordState
.TERMINATE_PENDING
or
701 self
._state
== VlRecordState
.FAILED
):
708 """ Terminate this VL """
709 if not self
.vlr_in_vns():
710 self
._log
.debug("Ignoring terminate request for id %s in state %s",
711 self
.id, self
._state
)
714 self
._log
.debug("Terminating VL id:%s", self
.id)
715 self
._state
= VlRecordState
.TERMINATE_PENDING
717 with self
._dts
.transaction(flags
=0) as xact
:
718 block
= xact
.block_create()
719 block
.add_query_delete(self
.xpath
)
720 yield from block
.execute(flags
=0, now
=True)
722 self
._state
= VlRecordState
.TERMINATED
723 self
._log
.debug("Terminated VL id:%s", self
.id)
726 class VnfRecordState(Enum
):
727 """ Vnf Record State """
729 INSTANTIATION_PENDING
= 102
731 TERMINATE_PENDING
= 104
736 class VirtualNetworkFunctionRecord(object):
737 """ Virtual Network Function Record class"""
738 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
742 def create_record(dts
, log
, loop
, project
, vnfd
, const_vnfd_msg
, nsd_id
, nsr_name
,
743 cloud_account_name
, om_datacenter_name
, nsr_id
, group_name
, group_instance_id
,
744 placement_groups
, restart_mode
=False):
745 """Creates a new VNFR object based on the given data.
747 If restart mode is enabled, then we look for existing records in the
748 DTS and create a VNFR records using the exiting data(ID)
751 VirtualNetworkFunctionRecord
753 vnfr_obj
= VirtualNetworkFunctionRecord(
768 restart_mode
=restart_mode
)
771 res_iter
= yield from dts
.query_read(
772 project
.add_project("D,/vnfr:vnfr-catalog/vnfr:vnfr"),
773 rwdts
.XactFlag
.MERGE
)
776 response
= yield from fut
777 vnfr
= response
.result
779 if vnfr
.name
== vnfr_obj
.name
:
780 vnfr_obj
.reset_id(vnfr
.id)
798 group_instance_id
=None,
799 placement_groups
= [],
800 restart_mode
= False):
804 self
._project
= project
806 self
._const
_vnfd
_msg
= const_vnfd_msg
807 self
._nsd
_id
= nsd_id
808 self
._nsr
_name
= nsr_name
809 self
._nsr
_id
= nsr_id
810 self
._cloud
_account
_name
= cloud_account_name
811 self
._om
_datacenter
_name
= om_datacenter_name
812 self
._group
_name
= group_name
813 self
._group
_instance
_id
= group_instance_id
814 self
._placement
_groups
= placement_groups
815 self
._config
_status
= NsrYang
.ConfigStates
.INIT
816 self
._create
_time
= int(time
.time())
818 self
._prev
_state
= VnfRecordState
.INIT
819 self
._state
= VnfRecordState
.INIT
820 self
._state
_failed
_reason
= None
822 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
825 self
._vnfr
_id
= str(uuid
.uuid4())
827 self
._vnfr
_msg
= self
.create_vnfr_msg()
828 self
._log
.debug("Set VNFR {} config type to {}".
829 format(self
.name
, self
.config_type
))
830 self
.restart_mode
= restart_mode
833 if group_name
is None and group_instance_id
is not None:
834 raise ValueError("Group instance id must not be provided with an empty group name")
844 return self
._project
.add_project("D,/vnfr:vnfr-catalog/vnfr:vnfr[vnfr:id = '{}']"
850 return self
._vnfr
_msg
853 def const_vnfr_msg(self
):
855 return RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr_ConstituentVnfrRef(
856 vnfr_id
=self
.id, cloud_account
=self
.cloud_account_name
,
857 om_datacenter
=self
._om
_datacenter
_name
)
865 def cloud_account_name(self
):
866 """ Cloud account that this VNF should be created in """
867 return self
._cloud
_account
_name
870 def om_datacenter_name(self
):
871 """ Datacenter that this VNF should be created in """
872 return self
._om
_datacenter
_name
877 """ Is this VNF actve """
878 return True if self
._state
== VnfRecordState
.ACTIVE
else False
882 """ state of this VNF """
886 def state_failed_reason(self
):
887 """ Error message in case this VNF is in failed state """
888 return self
._state
_failed
_reason
891 def member_vnf_index(self
):
892 """ Member VNF index """
893 return self
._const
_vnfd
_msg
.member_vnf_index
898 return self
._nsr
_name
902 """ Name of this VNFR """
903 if self
._name
is not None:
906 name_tags
= [self
._nsr
_name
]
908 if self
._group
_name
is not None:
909 name_tags
.append(self
._group
_name
)
911 if self
._group
_instance
_id
is not None:
912 name_tags
.append(str(self
._group
_instance
_id
))
914 name_tags
.extend([self
.vnfd
.name
, str(self
.member_vnf_index
)])
916 self
._name
= "__".join(name_tags
)
921 def vnfr_xpath(vnfr
):
922 """ Get the VNFR path from VNFR """
923 return (self
._project
.add_project(VirtualNetworkFunctionRecord
.XPATH
) +
924 "[vnfr:id = '{}']").format(vnfr
.id)
927 def config_type(self
):
928 cfg_types
= ['netconf', 'juju', 'script']
929 for method
in cfg_types
:
930 if self
._vnfd
.vnf_configuration
.has_field(method
):
935 def config_status(self
):
936 """Return the config status as YANG ENUM string"""
937 self
._log
.debug("Map VNFR {} config status {} ({})".
938 format(self
.name
, self
._config
_status
, self
.config_type
))
939 if self
.config_type
== 'none':
940 return 'config_not_needed'
941 elif self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
943 elif self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
948 def set_state(self
, state
):
949 """ set the state of this object """
950 self
._prev
_state
= self
._state
953 def reset_id(self
, vnfr_id
):
954 self
._vnfr
_id
= vnfr_id
955 self
._vnfr
_msg
= self
.create_vnfr_msg()
958 self
.config_store
.merge_vnfd_config(
961 self
.member_vnf_index
,
964 def create_vnfr_msg(self
):
965 """ VNFR message for this VNFR """
973 vnfd_copy_dict
= {k
: v
for k
, v
in self
._vnfd
.as_dict().items() if k
in vnfd_fields
}
976 "nsr_id_ref": self
._nsr
_id
,
978 "cloud_account": self
._cloud
_account
_name
,
979 "om_datacenter": self
._om
_datacenter
_name
,
980 "config_status": self
.config_status
982 vnfr_dict
.update(vnfd_copy_dict
)
984 vnfr
= RwVnfrYang
.YangData_RwProject_Project_VnfrCatalog_Vnfr
.from_dict(vnfr_dict
)
985 vnfr
.vnfd
= VnfrYang
.YangData_RwProject_Project_VnfrCatalog_Vnfr_Vnfd
. \
986 from_dict(self
.vnfd
.as_dict(),
987 ignore_missing_keys
=True)
988 vnfr
.member_vnf_index_ref
= self
.member_vnf_index
989 vnfr
.vnf_configuration
.from_dict(self
._vnfd
.vnf_configuration
.as_dict())
991 if self
._vnfd
.mgmt_interface
.has_field("port"):
992 vnfr
.mgmt_interface
.port
= self
._vnfd
.mgmt_interface
.port
994 for group_info
in self
._placement
_groups
:
995 group
= vnfr
.placement_groups_info
.add()
996 group
.from_dict(group_info
.as_dict())
998 # UI expects the monitoring param field to exist
999 vnfr
.monitoring_param
= []
1001 self
._log
.debug("Get vnfr_msg for VNFR {} : {}".format(self
.name
, vnfr
))
1005 def update_vnfm(self
):
1006 self
._log
.debug("Send an update to VNFM for VNFR {} with {}".
1007 format(self
.name
, self
.vnfr_msg
))
1008 yield from self
._dts
.query_update(
1010 0, #rwdts.XactFlag.TRACE,
1014 def get_config_status(self
):
1015 """Return the config status as YANG ENUM"""
1016 return self
._config
_status
1019 def set_config_status(self
, status
):
1021 def status_to_string(status
):
1023 NsrYang
.ConfigStates
.INIT
: 'init',
1024 NsrYang
.ConfigStates
.CONFIGURING
: 'configuring',
1025 NsrYang
.ConfigStates
.CONFIG_NOT_NEEDED
: 'config_not_needed',
1026 NsrYang
.ConfigStates
.CONFIGURED
: 'configured',
1027 NsrYang
.ConfigStates
.FAILED
: 'failed',
1030 return status_dc
[status
]
1032 self
._log
.debug("Update VNFR {} from {} ({}) to {}".
1033 format(self
.name
, self
._config
_status
,
1034 self
.config_type
, status
))
1035 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1036 self
._log
.error("Updating already configured VNFR {}".
1040 if self
._config
_status
!= status
:
1042 self
._config
_status
= status
1043 # I don't think this is used. Original implementor can check.
1044 # Caused Exception, so corrected it by status_to_string
1045 # But not sure whats the use of this variable?
1046 self
.vnfr_msg
.config_status
= status_to_string(status
)
1047 except Exception as e
:
1048 self
._log
.error("Exception=%s", str(e
))
1051 self
._log
.debug("Updated VNFR {} status to {}".format(self
.name
, status
))
1053 if self
._config
_status
!= NsrYang
.ConfigStates
.INIT
:
1055 # Publish only after VNFM has the VNFR created
1056 yield from self
.update_vnfm()
1057 except Exception as e
:
1058 self
._log
.error("Exception updating VNFM with new status {} of VNFR {}: {}".
1059 format(status
, self
.name
, e
))
1060 self
._log
.exception(e
)
1062 def is_configured(self
):
1063 if self
.config_type
== 'none':
1066 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1072 def instantiate(self
, nsr
):
1073 """ Instantiate this VNFR"""
1075 self
._log
.debug("Instaniating VNFR key %s, vnfd %s",
1076 self
.xpath
, self
._vnfd
)
1078 self
._log
.debug("Create VNF with xpath %s and vnfr %s",
1079 self
.xpath
, self
.vnfr_msg
)
1081 self
.set_state(VnfRecordState
.INSTANTIATION_PENDING
)
1083 def find_vlr_for_cp(conn
):
1084 """ Find VLR for the given connection point """
1085 for vlr
in nsr
.vlrs
:
1086 for vnfd_cp
in vlr
.vld_msg
.vnfd_connection_point_ref
:
1087 if (vnfd_cp
.vnfd_id_ref
== self
._vnfd
.id and
1088 vnfd_cp
.vnfd_connection_point_ref
== conn
.name
and
1089 vnfd_cp
.member_vnf_index_ref
== str(self
.member_vnf_index
) and
1090 vlr
.cloud_account_name
== self
.cloud_account_name
):
1091 self
._log
.debug("Found VLR for cp_name:%s and vnf-index:%d",
1092 conn
.name
, self
.member_vnf_index
)
1096 # For every connection point in the VNFD fill in the identifier
1097 self
._log
.debug("Add connection point for VNF %s: %s",
1098 self
.vnfr_msg
.name
, self
._vnfd
.connection_point
)
1099 for conn_p
in self
._vnfd
.connection_point
:
1100 cpr
= VnfrYang
.YangData_RwProject_Project_VnfrCatalog_Vnfr_ConnectionPoint()
1101 cpr
.name
= conn_p
.name
1102 cpr
.type_yang
= conn_p
.type_yang
1103 if conn_p
.has_field('port_security_enabled'):
1104 cpr
.port_security_enabled
= conn_p
.port_security_enabled
1106 vlr_ref
= find_vlr_for_cp(conn_p
)
1108 msg
= "Failed to find VLR for cp = %s" % conn_p
.name
1109 self
._log
.debug("%s", msg
)
1110 # raise VirtualNetworkFunctionRecordError(msg)
1113 cpr
.vlr_ref
= vlr_ref
.id
1114 self
.vnfr_msg
.connection_point
.append(cpr
)
1115 self
._log
.debug("Connection point [%s] added, vnf id=%s vnfd id=%s",
1116 cpr
, self
.vnfr_msg
.id, self
.vnfr_msg
.vnfd
.id)
1118 if not self
.restart_mode
:
1119 yield from self
._dts
.query_create(self
.xpath
,
1123 yield from self
._dts
.query_update(self
.xpath
,
1127 self
._log
.info("Created VNF with xpath %s and vnfr %s",
1128 self
.xpath
, self
.vnfr_msg
)
1131 def update_state(self
, vnfr_msg
):
1132 """ Update this VNFR"""
1133 if vnfr_msg
.operational_status
== "running":
1134 if self
.vnfr_msg
.operational_status
!= "running":
1135 yield from self
.is_active()
1136 elif vnfr_msg
.operational_status
== "failed":
1137 yield from self
.instantiation_failed(failed_reason
=vnfr_msg
.operational_status_details
)
1140 def is_active(self
):
1141 """ This VNFR is active """
1142 self
._log
.debug("VNFR %s is active", self
._vnfr
_id
)
1143 self
.set_state(VnfRecordState
.ACTIVE
)
1146 def instantiation_failed(self
, failed_reason
=None):
1147 """ This VNFR instantiation failed"""
1148 self
._log
.error("VNFR %s instantiation failed", self
._vnfr
_id
)
1149 self
.set_state(VnfRecordState
.FAILED
)
1150 self
._state
_failed
_reason
= failed_reason
1152 def vnfr_in_vnfm(self
):
1153 """ Is there a VNFR record in VNFM """
1154 if (self
._state
== VnfRecordState
.ACTIVE
or
1155 self
._state
== VnfRecordState
.INSTANTIATION_PENDING
or
1156 self
._state
== VnfRecordState
.FAILED
):
1162 def terminate(self
):
1163 """ Terminate this VNF """
1164 if not self
.vnfr_in_vnfm():
1165 self
._log
.debug("Ignoring terminate request for id %s in state %s",
1166 self
.id, self
._state
)
1169 self
._log
.debug("Terminating VNF id:%s", self
.id)
1170 self
.set_state(VnfRecordState
.TERMINATE_PENDING
)
1171 with self
._dts
.transaction(flags
=0) as xact
:
1172 block
= xact
.block_create()
1173 block
.add_query_delete(self
.xpath
)
1174 yield from block
.execute(flags
=0)
1175 self
.set_state(VnfRecordState
.TERMINATED
)
1176 self
._log
.debug("Terminated VNF id:%s", self
.id)
1179 class NetworkServiceStatus(object):
1180 """ A class representing the Network service's status """
1181 MAX_EVENTS_RECORDED
= 10
1182 """ Network service Status class"""
1183 def __init__(self
, dts
, log
, loop
):
1188 self
._state
= NetworkServiceRecordState
.INIT
1189 self
._events
= deque([])
1192 def create_notification(self
, evt
, evt_desc
, evt_details
):
1193 xp
= "N,/rw-nsr:nsm-notification"
1194 notif
= RwNsrYang
.YangNotif_RwNsr_NsmNotification()
1196 notif
.description
= evt_desc
1197 notif
.details
= evt_details
if evt_details
is not None else None
1199 yield from self
._dts
.query_create(xp
, rwdts
.XactFlag
.ADVISE
, notif
)
1200 self
._log
.info("Notification called by creating dts query: %s", notif
)
1202 def record_event(self
, evt
, evt_desc
, evt_details
):
1203 """ Record an event """
1204 self
._log
.debug("Recording event - evt %s, evt_descr %s len = %s",
1205 evt
, evt_desc
, len(self
._events
))
1206 if len(self
._events
) >= NetworkServiceStatus
.MAX_EVENTS_RECORDED
:
1207 self
._events
.popleft()
1208 self
._events
.append((int(time
.time()), evt
, evt_desc
,
1209 evt_details
if evt_details
is not None else None))
1211 self
._loop
.create_task(self
.create_notification(evt
,evt_desc
,evt_details
))
1213 def set_state(self
, state
):
1214 """ set the state of this status object """
1218 """ Return the state as a yang enum string """
1219 state_to_str_map
= {"INIT": "init",
1220 "VL_INIT_PHASE": "vl_init_phase",
1221 "VNF_INIT_PHASE": "vnf_init_phase",
1222 "VNFFG_INIT_PHASE": "vnffg_init_phase",
1223 "SCALING_GROUP_INIT_PHASE": "scaling_group_init_phase",
1224 "RUNNING": "running",
1225 "SCALING_OUT": "scaling_out",
1226 "SCALING_IN": "scaling_in",
1227 "TERMINATE_RCVD": "terminate_rcvd",
1228 "TERMINATE": "terminate",
1229 "VL_TERMINATE_PHASE": "vl_terminate_phase",
1230 "VNF_TERMINATE_PHASE": "vnf_terminate_phase",
1231 "VNFFG_TERMINATE_PHASE": "vnffg_terminate_phase",
1232 "TERMINATED": "terminated",
1234 "VL_INSTANTIATE": "vl_instantiate",
1235 "VL_TERMINATE": "vl_terminate",
1237 return state_to_str_map
[self
._state
.name
]
1241 """ State of this status object """
1246 """ Network Service Record as a message"""
1249 for entry
in self
._events
:
1250 event
= RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr_OperationalEvents()
1253 event
.timestamp
, event
.event
, event
.description
, event
.details
= entry
1254 event_list
.append(event
)
1258 class NetworkServiceRecord(object):
1259 """ Network service record """
1260 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
1262 def __init__(self
, dts
, log
, loop
, nsm
, nsm_plugin
, nsr_cfg_msg
,
1263 sdn_account_name
, key_pairs
, project
, restart_mode
=False,
1269 self
._nsr
_cfg
_msg
= nsr_cfg_msg
1270 self
._nsm
_plugin
= nsm_plugin
1271 self
._sdn
_account
_name
= sdn_account_name
1272 self
._vlr
_handler
= vlr_handler
1273 self
._project
= project
1276 self
._nsr
_msg
= None
1277 self
._nsr
_regh
= None
1278 self
._key
_pairs
= key_pairs
1283 self
._param
_pools
= {}
1284 self
._scaling
_groups
= {}
1285 self
._create
_time
= int(time
.time())
1286 self
._op
_status
= NetworkServiceStatus(dts
, log
, loop
)
1287 self
._config
_status
= NsrYang
.ConfigStates
.CONFIGURING
1288 self
._config
_status
_details
= None
1290 self
.restart_mode
= restart_mode
1291 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
1292 self
._debug
_running
= False
1293 self
._is
_active
= False
1294 self
._vl
_phase
_completed
= False
1295 self
._vnf
_phase
_completed
= False
1296 self
.vlr_uptime_tasks
= {}
1299 # Initalise the state to init
1300 # The NSR moves through the following transitions
1301 # 1. INIT -> VLS_READY once all the VLs in the NSD are created
1302 # 2. VLS_READY - VNFS_READY when all the VNFs in the NSD are created
1303 # 3. VNFS_READY - READY when the NSR is published
1305 self
.set_state(NetworkServiceRecordState
.INIT
)
1307 self
.substitute_input_parameters
= InputParameterSubstitution(self
._log
, self
._project
)
1310 def nsm_plugin(self
):
1312 return self
._nsm
_plugin
1314 def set_state(self
, state
):
1315 """ Set state for this NSR"""
1316 self
._log
.debug("Setting state to %s", state
)
1317 # We are in init phase and is moving to the next state
1318 # The new state could be a FAILED state or VNF_INIIT_PHASE
1319 if self
.state
== NetworkServiceRecordState
.VL_INIT_PHASE
:
1320 self
._vl
_phase
_completed
= True
1322 if self
.state
== NetworkServiceRecordState
.VNF_INIT_PHASE
:
1323 self
._vnf
_phase
_completed
= True
1325 self
._op
_status
.set_state(state
)
1326 self
._nsm
_plugin
.set_state(self
.id, state
)
1330 """ Get id for this NSR"""
1331 return self
._nsr
_cfg
_msg
.id
1335 """ Name of this network service record """
1336 return self
._nsr
_cfg
_msg
.name
1339 def cloud_account_name(self
):
1340 return self
._nsr
_cfg
_msg
.cloud_account
1343 def om_datacenter_name(self
):
1344 if self
._nsr
_cfg
_msg
.has_field('om_datacenter'):
1345 return self
._nsr
_cfg
_msg
.om_datacenter
1350 """State of this NetworkServiceRecord"""
1351 return self
._op
_status
.state
1355 """ Is this NSR active ?"""
1356 return True if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
else False
1360 """ VLRs associated with this NSR"""
1365 """ VNFRs associated with this NSR"""
1370 """ VNFFGRs associated with this NSR"""
1371 return self
._vnffgrs
1374 def scaling_groups(self
):
1375 """ Scaling groups associated with this NSR """
1376 return self
._scaling
_groups
1379 def param_pools(self
):
1380 """ Parameter value pools associated with this NSR"""
1381 return self
._param
_pools
1384 def nsr_cfg_msg(self
):
1385 return self
._nsr
_cfg
_msg
1388 def nsr_cfg_msg(self
, msg
):
1389 self
._nsr
_cfg
_msg
= msg
1393 """ NSD Protobuf for this NSR """
1394 if self
._nsd
is not None:
1396 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
1401 """ NSD ID for this NSR """
1402 return self
.nsd_msg
.id
1406 ''' Get a new job id for config primitive'''
1411 def config_status(self
):
1412 """ Config status for NSR """
1413 return self
._config
_status
1415 def resolve_placement_group_cloud_construct(self
, input_group
):
1417 Returns the cloud specific construct for placement group
1419 copy_dict
= ['name', 'requirement', 'strategy']
1421 for group_info
in self
._nsr
_cfg
_msg
.nsd_placement_group_maps
:
1422 if group_info
.placement_group_ref
== input_group
.name
:
1423 group
= VnfrYang
.YangData_RwProject_Project_VnfrCatalog_Vnfr_PlacementGroupsInfo()
1424 group_dict
= {k
:v
for k
,v
in
1425 group_info
.as_dict().items() if k
!= 'placement_group_ref'}
1426 for param
in copy_dict
:
1427 group_dict
.update({param
: getattr(input_group
, param
)})
1428 group
.from_dict(group_dict
)
1434 return "NSR(name={}, nsd_id={}, cloud_account={})".format(
1435 self
.name
, self
.nsd_id
, self
.cloud_account_name
1438 def _get_vnfd(self
, vnfd_id
, config_xact
):
1439 """ Fetch vnfd msg for the passed vnfd id """
1440 return self
._nsm
.get_vnfd(vnfd_id
, config_xact
)
1442 def _get_vnfd_cloud_account(self
, vnfd_member_index
):
1443 """ Fetch Cloud Account for the passed vnfd id """
1444 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1445 vim_accounts
= [(vnf
.cloud_account
,vnf
.om_datacenter
) for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map \
1446 if str(vnfd_member_index
) == vnf
.member_vnf_index_ref
]
1447 if vim_accounts
and vim_accounts
[0]:
1448 return vim_accounts
[0]
1449 return (self
.cloud_account_name
,self
.om_datacenter_name
)
1451 def _get_constituent_vnfd_msg(self
, vnf_index
):
1452 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1453 if const_vnfd
.member_vnf_index
== vnf_index
:
1456 raise ValueError("Constituent VNF index %s not found" % vnf_index
)
1458 def record_event(self
, evt
, evt_desc
, evt_details
=None, state
=None):
1459 """ Record an event """
1460 self
._op
_status
.record_event(evt
, evt_desc
, evt_details
)
1461 if state
is not None:
1462 self
.set_state(state
)
1464 def scaling_trigger_str(self
, trigger
):
1465 SCALING_TRIGGER_STRS
= {
1466 NsdYang
.ScalingTrigger
.PRE_SCALE_IN
: 'pre-scale-in',
1467 NsdYang
.ScalingTrigger
.POST_SCALE_IN
: 'post-scale-in',
1468 NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
: 'pre-scale-out',
1469 NsdYang
.ScalingTrigger
.POST_SCALE_OUT
: 'post-scale-out',
1472 return SCALING_TRIGGER_STRS
[trigger
]
1473 except Exception as e
:
1474 self
._log
.error("Scaling trigger mapping error for {} : {}".
1476 self
._log
.exception(e
)
1477 return "Unknown trigger"
1480 def instantiate_vls(self
):
1482 This function instantiates VLs for every VL in this Network Service
1484 self
._log
.debug("Instantiating %d VLs in NSD id %s", len(self
._vlrs
),
1486 for vlr
in self
._vlrs
:
1487 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1488 vlr
.state
= VlRecordState
.ACTIVE
1489 self
.vlr_uptime_tasks
[vlr
.id] = self
._loop
.create_task(self
.vlr_uptime_update(vlr
))
1492 def vlr_uptime_update(self
, vlr
):
1495 vlr_
= RwVlrYang
.YangData_RwProject_Project_VlrCatalog_Vlr
.from_dict({'id': vlr
.id})
1497 vlr_
.uptime
= int(time
.time()) - vlr
._create
_time
1498 xpath
= self
._project
.add_project(VirtualLinkRecord
.vlr_xpath(vlr
))
1499 yield from self
._vlr
_handler
.update(None, xpath
, vlr_
)
1500 yield from asyncio
.sleep(2, loop
=self
._loop
)
1501 except asyncio
.CancelledError
:
1502 self
._log
.debug("Received cancellation request for vlr_uptime_update task")
1503 xpath
= self
._project
.add_project(VirtualLinkRecord
.vlr_xpath(vlr
))
1504 yield from self
._vlr
_handler
.delete(None, xpath
)
1508 def create(self
, config_xact
):
1509 """ Create this network service"""
1510 # Create virtual links for all the external vnf
1511 # connection points in this NS
1512 yield from self
.create_vls()
1514 # Create VNFs in this network service
1515 yield from self
.create_vnfs(config_xact
)
1517 # Create VNFFG for network service
1518 self
.create_vnffgs()
1520 # Create Scaling Groups for each scaling group in NSD
1521 self
.create_scaling_groups()
1523 # Create Parameter Pools
1524 self
.create_param_pools()
1527 def apply_scale_group_config_script(self
, script
, group
, scale_instance
, trigger
, vnfrs
=None):
1528 """ Apply config based on script for scale group """
1531 def add_vnfrs_data(vnfrs_list
):
1532 """ Add as a dict each of the VNFRs data """
1534 for vnfr
in vnfrs_list
:
1535 self
._log
.debug("Add VNFR {} data".format(vnfr
))
1537 vnfr_data
['name'] = vnfr
.name
1538 if trigger
in [NsdYang
.ScalingTrigger
.PRE_SCALE_IN
, NsdYang
.ScalingTrigger
.POST_SCALE_OUT
]:
1539 # Get VNF management and other IPs, etc
1540 opdata
= yield from self
.fetch_vnfr(vnfr
.xpath
)
1541 self
._log
.debug("VNFR {} op data: {}".format(vnfr
.name
, opdata
))
1543 vnfr_data
['rw_mgmt_ip'] = opdata
.mgmt_interface
.ip_address
1544 vnfr_data
['rw_mgmt_port'] = opdata
.mgmt_interface
.port
1545 except Exception as e
:
1546 self
._log
.error("Unable to get management IP for vnfr {}:{}".
1547 format(vnfr
.name
, e
))
1550 vnfr_data
['connection_points'] = []
1551 for cp
in opdata
.connection_point
:
1553 con_pt
['name'] = cp
.name
1554 con_pt
['ip_address'] = cp
.ip_address
1555 vnfr_data
['connection_points'].append(con_pt
)
1556 except Exception as e
:
1557 self
._log
.error("Exception getting connections points for VNFR {}: {}".
1558 format(vnfr
.name
, e
))
1560 vnfrs_data
.append(vnfr_data
)
1561 self
._log
.debug("VNFRs data: {}".format(vnfrs_data
))
1565 def add_nsr_data(nsr
):
1567 nsr_data
['name'] = nsr
.name
1570 if script
is None or len(script
) == 0:
1571 self
._log
.error("Script not provided for scale group config: {}".format(group
.name
))
1574 if script
[0] == '/':
1577 path
= os
.path
.join(os
.environ
['RIFT_INSTALL'], "usr/bin", script
)
1578 if not os
.path
.exists(path
):
1579 self
._log
.error("Config faled for scale group {}: Script does not exist at {}".
1580 format(group
.name
, path
))
1583 # Build a YAML file with all parameters for the script to execute
1584 # The data consists of 5 sections
1586 # 2. Scale group config
1587 # 3. VNFRs in the scale group
1588 # 4. VNFRs outside scale group
1591 data
['trigger'] = group
.trigger_map(trigger
)
1592 data
['config'] = group
.group_msg
.as_dict()
1595 data
["vnfrs_in_group"] = yield from add_vnfrs_data(vnfrs
)
1597 data
["vnfrs_in_group"] = yield from add_vnfrs_data(scale_instance
.vnfrs
)
1599 data
["vnfrs_others"] = yield from add_vnfrs_data(self
.vnfrs
.values())
1600 data
["nsr"] = add_nsr_data(self
)
1603 with tempfile
.NamedTemporaryFile(delete
=False) as tmp_file
:
1604 tmp_file
.write(yaml
.dump(data
, default_flow_style
=True)
1607 self
._log
.debug("Creating a temp file: {} with input data: {}".
1608 format(tmp_file
.name
, data
))
1610 cmd
= "{} {}".format(path
, tmp_file
.name
)
1611 self
._log
.debug("Running the CMD: {}".format(cmd
))
1612 proc
= yield from asyncio
.create_subprocess_shell(cmd
, loop
=self
._loop
)
1613 rc
= yield from proc
.wait()
1615 self
._log
.error("The script {} for scale group {} config returned: {}".
1616 format(script
, group
.name
, rc
))
1624 def apply_scaling_group_config(self
, trigger
, group
, scale_instance
, vnfrs
=None):
1625 """ Apply the config for the scaling group based on trigger """
1626 if group
is None or scale_instance
is None:
1630 def update_config_status(success
=True, err_msg
=None):
1631 self
._log
.debug("Update %s config status to %r : %s",
1632 scale_instance
, success
, err_msg
)
1633 if (scale_instance
.config_status
== "failed"):
1634 # Do not update the config status if it is already in failed state
1637 if scale_instance
.config_status
== "configured":
1638 # Update only to failed state an already configured scale instance
1640 scale_instance
.config_status
= "failed"
1641 scale_instance
.config_err_msg
= err_msg
1642 yield from self
.update_state()
1644 # We are in configuring state
1645 # Only after post scale out mark instance as configured
1646 if trigger
== NsdYang
.ScalingTrigger
.POST_SCALE_OUT
:
1648 scale_instance
.config_status
= "configured"
1650 scale_instance
.config_status
= "failed"
1651 scale_instance
.config_err_msg
= err_msg
1652 yield from self
.update_state()
1654 config
= group
.trigger_config(trigger
)
1658 self
._log
.debug("Scaling group {} config: {}".format(group
.name
, config
))
1659 if config
.has_field("ns_config_primitive_name_ref"):
1660 config_name
= config
.ns_config_primitive_name_ref
1661 nsd_msg
= self
.nsd_msg
1662 config_primitive
= None
1663 for ns_cfg_prim
in nsd_msg
.service_primitive
:
1664 if ns_cfg_prim
.name
== config_name
:
1665 config_primitive
= ns_cfg_prim
1668 if config_primitive
is None:
1669 raise ValueError("Could not find ns_cfg_prim %s in nsr %s" % (config_name
, self
.name
))
1671 self
._log
.debug("Scaling group {} config primitive: {}".format(group
.name
, config_primitive
))
1672 if config_primitive
.has_field("user_defined_script"):
1673 rc
= yield from self
.apply_scale_group_config_script(config_primitive
.user_defined_script
,
1674 group
, scale_instance
, trigger
, vnfrs
)
1677 err_msg
= "Failed config for trigger {} using config script '{}'". \
1678 format(self
.scaling_trigger_str(trigger
),
1679 config_primitive
.user_defined_script
)
1680 yield from update_config_status(success
=rc
, err_msg
=err_msg
)
1683 err_msg
= "Failed config for trigger {} as config script is not specified". \
1684 format(self
.scaling_trigger_str(trigger
))
1685 yield from update_config_status(success
=False, err_msg
=err_msg
)
1686 raise NotImplementedError("Only script based config support for scale group for now: {}".
1689 err_msg
= "Failed config for trigger {} as config primitive is not specified".\
1690 format(self
.scaling_trigger_str(trigger
))
1691 yield from update_config_status(success
=False, err_msg
=err_msg
)
1692 self
._log
.error("Config primitive not specified for config action in scale group %s" %
1696 def create_scaling_groups(self
):
1697 """ This function creates a NSScalingGroup for every scaling
1698 group defined in he NSD"""
1700 for scaling_group_msg
in self
.nsd_msg
.scaling_group_descriptor
:
1701 self
._log
.debug("Found scaling_group %s in nsr id %s",
1702 scaling_group_msg
.name
, self
.id)
1704 group_record
= scale_group
.ScalingGroup(
1709 self
._scaling
_groups
[group_record
.name
] = group_record
1712 def create_scale_group_instance(self
, group_name
, index
, config_xact
, is_default
=False):
1713 group
= self
._scaling
_groups
[group_name
]
1714 scale_instance
= group
.create_instance(index
, is_default
)
1718 self
._log
.debug("Creating %u VNFs associated with NS id %s scaling group %s",
1719 len(self
.nsd_msg
.constituent_vnfd
), self
.id, self
)
1722 for vnf_index
, count
in group
.vnf_index_count_map
.items():
1723 const_vnfd_msg
= self
._get
_constituent
_vnfd
_msg
(vnf_index
)
1724 vnfd_msg
= self
._get
_vnfd
(const_vnfd_msg
.vnfd_id_ref
, config_xact
)
1726 cloud_account_name
, om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd_msg
.member_vnf_index
)
1727 if cloud_account_name
is None:
1728 cloud_account_name
= self
.cloud_account_name
1729 for _
in range(count
):
1730 vnfr
= yield from self
.create_vnf_record(vnfd_msg
, const_vnfd_msg
, cloud_account_name
, om_datacenter_name
, group_name
, index
)
1731 scale_instance
.add_vnfr(vnfr
)
1737 def instantiate_instance():
1738 self
._log
.debug("Creating %s VNFRS", scale_instance
)
1739 vnfrs
= yield from create_vnfs()
1740 yield from self
.publish()
1742 self
._log
.debug("Instantiating %s VNFRS for %s", len(vnfrs
), scale_instance
)
1743 scale_instance
.operational_status
= "vnf_init_phase"
1744 yield from self
.update_state()
1747 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
,
1748 group
, scale_instance
, vnfrs
)
1750 self
._log
.error("Pre scale out config for scale group {} ({}) failed".
1751 format(group
.name
, index
))
1752 scale_instance
.operational_status
= "failed"
1754 yield from self
.instantiate_vnfs(vnfrs
)
1756 except Exception as e
:
1757 self
._log
.exception("Failed to begin instantiatiation of vnfs for scale group {}: {}".
1758 format(group
.name
, e
))
1759 self
._log
.exception(e
)
1760 scale_instance
.operational_status
= "failed"
1762 yield from self
.update_state()
1764 yield from instantiate_instance()
1767 def delete_scale_group_instance(self
, group_name
, index
):
1768 group
= self
._scaling
_groups
[group_name
]
1769 scale_instance
= group
.get_instance(index
)
1770 if scale_instance
.is_default
:
1771 raise ScalingOperationError("Cannot terminate a default scaling group instance")
1773 scale_instance
.operational_status
= "terminate"
1774 yield from self
.update_state()
1777 def terminate_instance():
1778 self
._log
.debug("Terminating %s VNFRS" % scale_instance
)
1779 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_IN
,
1780 group
, scale_instance
)
1782 self
._log
.error("Pre scale in config for scale group {} ({}) failed".
1783 format(group
.name
, index
))
1785 # Going ahead with terminate, even if there is an error in pre-scale-in config
1786 # as this could be result of scale out failure and we need to cleanup this group
1787 yield from self
.terminate_vnfrs(scale_instance
.vnfrs
)
1788 group
.delete_instance(index
)
1790 scale_instance
.operational_status
= "vnf_terminate_phase"
1791 yield from self
.update_state()
1793 yield from terminate_instance()
1796 def _update_scale_group_instances_status(self
):
1798 def post_scale_out_task(group
, instance
):
1799 # Apply post scale out config once all VNFRs are active
1800 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_OUT
,
1802 instance
.operational_status
= "running"
1804 self
._log
.debug("Scale out for group {} and instance {} succeeded".
1805 format(group
.name
, instance
.instance_id
))
1807 self
._log
.error("Post scale out config for scale group {} ({}) failed".
1808 format(group
.name
, instance
.instance_id
))
1810 yield from self
.update_state()
1812 group_instances
= {group
: group
.instances
for group
in self
._scaling
_groups
.values()}
1813 for group
, instances
in group_instances
.items():
1814 self
._log
.debug("Updating %s instance status", group
)
1815 for instance
in instances
:
1816 instance_vnf_state_list
= [vnfr
.state
for vnfr
in instance
.vnfrs
]
1817 self
._log
.debug("Got vnfr instance states: %s", instance_vnf_state_list
)
1818 if instance
.operational_status
== "vnf_init_phase":
1819 if all([state
== VnfRecordState
.ACTIVE
for state
in instance_vnf_state_list
]):
1820 instance
.operational_status
= "running"
1822 # Create a task for post scale out to allow us to sleep before attempting
1823 # to configure newly created VM's
1824 self
._loop
.create_task(post_scale_out_task(group
, instance
))
1826 elif any([state
== VnfRecordState
.FAILED
for state
in instance_vnf_state_list
]):
1827 self
._log
.debug("Scale out for group {} and instance {} failed".
1828 format(group
.name
, instance
.instance_id
))
1829 instance
.operational_status
= "failed"
1831 elif instance
.operational_status
== "vnf_terminate_phase":
1832 if all([state
== VnfRecordState
.TERMINATED
for state
in instance_vnf_state_list
]):
1833 instance
.operational_status
= "terminated"
1834 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_IN
,
1837 self
._log
.debug("Scale in for group {} and instance {} succeeded".
1838 format(group
.name
, instance
.instance_id
))
1840 self
._log
.error("Post scale in config for scale group {} ({}) failed".
1841 format(group
.name
, instance
.instance_id
))
1843 def create_vnffgs(self
):
1844 """ This function creates VNFFGs for every VNFFG in the NSD
1845 associated with this NSR"""
1847 for vnffgd
in self
.nsd_msg
.vnffgd
:
1848 self
._log
.debug("Found vnffgd %s in nsr id %s", vnffgd
, self
.id)
1849 vnffgr
= VnffgRecord(self
._dts
,
1852 self
._nsm
._vnffgmgr
,
1856 self
._sdn
_account
_name
1858 self
._vnffgrs
[vnffgr
.id] = vnffgr
1860 def resolve_vld_ip_profile(self
, nsd_msg
, vld
):
1861 self
._log
.debug("Receieved ip profile ref is %s",vld
.ip_profile_ref
)
1862 if not vld
.has_field('ip_profile_ref'):
1864 profile
= [profile
for profile
in nsd_msg
.ip_profiles
if profile
.name
== vld
.ip_profile_ref
]
1865 return profile
[0] if profile
else None
1868 def _create_vls(self
, vld
, cloud_account
,om_datacenter
):
1869 """Create a VLR in the cloud account specified using the given VLD
1873 cloud_account : Cloud account name
1878 vlr
= yield from VirtualLinkRecord
.create_record(
1887 self
.resolve_vld_ip_profile(self
.nsd_msg
, vld
),
1889 restart_mode
=self
.restart_mode
)
1893 def _extract_cloud_accounts_for_vl(self
, vld
):
1895 Extracts the list of cloud accounts from the NS Config obj
1898 1. Cloud accounts based connection point (vnf_cloud_account_map)
1900 vld : VLD yang object
1905 cloud_account_list
= []
1907 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1908 # Handle case where cloud_account is None
1910 for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1911 if vnf
.cloud_account
is not None or vnf
.om_datacenter
is not None:
1912 vnf_cloud_map
[vnf
.member_vnf_index_ref
] = (vnf
.cloud_account
,vnf
.om_datacenter
)
1914 for vnfc
in vld
.vnfd_connection_point_ref
:
1915 cloud_account
= vnf_cloud_map
.get(
1916 vnfc
.member_vnf_index_ref
,
1917 (self
.cloud_account_name
,self
.om_datacenter_name
))
1919 cloud_account_list
.append(cloud_account
)
1921 if self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1922 for vld_map
in self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1923 if vld_map
.vld_id_ref
== vld
.id:
1924 for cloud_account
in vld_map
.cloud_accounts
:
1925 cloud_account_list
.extend((cloud_account
,None))
1926 for om_datacenter
in vld_map
.om_datacenters
:
1927 cloud_account_list
.extend((None,om_datacenter
))
1929 # If no config has been provided then fall-back to the default
1931 if not cloud_account_list
:
1932 cloud_account_list
= [(self
.cloud_account_name
,self
.om_datacenter_name
)]
1934 self
._log
.debug("VL {} cloud accounts: {}".
1935 format(vld
.name
, cloud_account_list
))
1936 return set(cloud_account_list
)
1939 def create_vls(self
):
1940 """ This function creates VLs for every VLD in the NSD
1941 associated with this NSR"""
1942 for vld
in self
.nsd_msg
.vld
:
1944 self
._log
.debug("Found vld %s in nsr id %s", vld
, self
.id)
1945 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1946 for cloud_account
,om_datacenter
in cloud_account_list
:
1947 vlr
= yield from self
._create
_vls
(vld
, cloud_account
,om_datacenter
)
1948 self
._vlrs
.append(vlr
)
1952 def create_vl_instance(self
, vld
):
1953 self
._log
.debug("Create VL for {}: {}".format(self
.id, vld
.as_dict()))
1954 # Check if the VL is already present
1956 for vl
in self
._vlrs
:
1957 if vl
.vld_msg
.id == vld
.id:
1958 self
._log
.debug("The VLD %s already in NSR %s as VLR %s with status %s",
1959 vld
.id, self
.id, vl
.id, vl
.state
)
1961 if vlr
.state
!= VlRecordState
.TERMINATED
:
1962 err_msg
= "VLR for VL %s in NSR %s already instantiated", \
1964 self
._log
.error(err_msg
)
1965 raise NsrVlUpdateError(err_msg
)
1969 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1970 for account
,om_datacenter
in cloud_account_list
:
1971 vlr
= yield from self
._create
_vls
(vld
, account
,om_datacenter
)
1972 self
._vlrs
.append(vlr
)
1974 vlr
.state
= VlRecordState
.INSTANTIATION_PENDING
1975 yield from self
.update_state()
1978 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1979 vlr
.state
= VlRecordState
.ACTIVE
1981 except Exception as e
:
1982 err_msg
= "Error instantiating VL for NSR {} and VLD {}: {}". \
1983 format(self
.id, vld
.id, e
)
1984 self
._log
.error(err_msg
)
1985 self
._log
.exception(e
)
1986 vlr
.state
= VlRecordState
.FAILED
1988 yield from self
.update_state()
1991 def delete_vl_instance(self
, vld
):
1992 for vlr
in self
._vlrs
:
1993 if vlr
.vld_msg
.id == vld
.id:
1994 self
._log
.debug("Found VLR %s for VLD %s in NSR %s",
1995 vlr
.id, vld
.id, self
.id)
1996 vlr
.state
= VlRecordState
.TERMINATE_PENDING
1997 yield from self
.update_state()
2000 yield from self
.nsm_plugin
.terminate_vl(vlr
)
2001 vlr
.state
= VlRecordState
.TERMINATED
2002 self
._vlrs
.remove(vlr
)
2004 except Exception as e
:
2005 err_msg
= "Error terminating VL for NSR {} and VLD {}: {}". \
2006 format(self
.id, vld
.id, e
)
2007 self
._log
.error(err_msg
)
2008 self
._log
.exception(e
)
2009 vlr
.state
= VlRecordState
.FAILED
2011 yield from self
.update_state()
2015 def create_vnfs(self
, config_xact
):
2017 This function creates VNFs for every VNF in the NSD
2018 associated with this NSR
2020 self
._log
.debug("Creating %u VNFs associated with this NS id %s",
2021 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
2023 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
2024 if not const_vnfd
.start_by_default
:
2025 self
._log
.debug("start_by_default set to False in constituent VNF (%s). Skipping start.",
2026 const_vnfd
.member_vnf_index
)
2029 vnfd_msg
= self
._get
_vnfd
(const_vnfd
.vnfd_id_ref
, config_xact
)
2030 cloud_account_name
,om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd
.member_vnf_index
)
2031 if cloud_account_name
is None:
2032 cloud_account_name
= self
.cloud_account_name
2033 yield from self
.create_vnf_record(vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
)
2036 def get_placement_groups(self
, vnfd_msg
, const_vnfd
):
2037 placement_groups
= []
2038 for group
in self
.nsd_msg
.placement_groups
:
2039 for member_vnfd
in group
.member_vnfd
:
2040 if (member_vnfd
.vnfd_id_ref
== vnfd_msg
.id) and \
2041 (member_vnfd
.member_vnf_index_ref
== str(const_vnfd
.member_vnf_index
)):
2042 group_info
= self
.resolve_placement_group_cloud_construct(group
)
2043 if group_info
is None:
2044 self
._log
.error("Could not resolve cloud-construct for placement group: %s", group
.name
)
2045 ### raise PlacementGroupError("Could not resolve cloud-construct for placement group: {}".format(group.name))
2047 self
._log
.info("Successfully resolved cloud construct for placement group: %s for VNF: %s (Member Index: %s)",
2050 const_vnfd
.member_vnf_index
)
2051 placement_groups
.append(group_info
)
2052 return placement_groups
2055 def create_vnf_record(self
, vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
, group_name
=None, group_instance_id
=None):
2056 # Fetch the VNFD associated with this VNF
2057 placement_groups
= self
.get_placement_groups(vnfd_msg
, const_vnfd
)
2058 self
._log
.info("Cloud Account for VNF %d is %s",const_vnfd
.member_vnf_index
,cloud_account_name
)
2059 self
._log
.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s",
2061 const_vnfd
.member_vnf_index
,
2062 [ group
.name
for group
in placement_groups
])
2063 vnfr
= yield from VirtualNetworkFunctionRecord
.create_record(self
._dts
,
2077 restart_mode
=self
.restart_mode
,
2079 if vnfr
.id in self
._vnfrs
:
2080 err
= "VNF with VNFR id %s already in vnf list" % (vnfr
.id,)
2081 raise NetworkServiceRecordError(err
)
2083 self
._vnfrs
[vnfr
.id] = vnfr
2084 self
._nsm
.vnfrs
[vnfr
.id] = vnfr
2086 yield from vnfr
.set_config_status(NsrYang
.ConfigStates
.INIT
)
2088 self
._log
.debug("Added VNFR %s to NSM VNFR list with id %s",
2094 def create_param_pools(self
):
2095 for param_pool
in self
.nsd_msg
.parameter_pool
:
2096 self
._log
.debug("Found parameter pool %s in nsr id %s", param_pool
, self
.id)
2098 start_value
= param_pool
.range.start_value
2099 end_value
= param_pool
.range.end_value
2100 if end_value
< start_value
:
2101 raise NetworkServiceRecordError(
2102 "Parameter pool %s has invalid range (start: {}, end: {})".format(
2103 start_value
, end_value
2107 self
._param
_pools
[param_pool
.name
] = config_value_pool
.ParameterValuePool(
2110 range(start_value
, end_value
)
2114 def fetch_vnfr(self
, vnfr_path
):
2115 """ Fetch VNFR record """
2117 self
._log
.debug("Fetching VNFR with key %s while instantiating %s",
2119 res_iter
= yield from self
._dts
.query_read(vnfr_path
, rwdts
.XactFlag
.MERGE
)
2121 for ent
in res_iter
:
2122 res
= yield from ent
2128 def instantiate_vnfs(self
, vnfrs
):
2130 This function instantiates VNFs for every VNF in this Network Service
2132 self
._log
.debug("Instantiating %u VNFs in NS %s", len(vnfrs
), self
.id)
2134 self
._log
.debug("Instantiating VNF: %s in NS %s", vnf
, self
.id)
2135 yield from self
.nsm_plugin
.instantiate_vnf(self
, vnf
)
2138 def instantiate_vnffgs(self
):
2140 This function instantiates VNFFGs for every VNFFG in this Network Service
2142 self
._log
.debug("Instantiating %u VNFFGs in NS %s",
2143 len(self
.nsd_msg
.vnffgd
), self
.id)
2144 for _
, vnfr
in self
.vnfrs
.items():
2145 while vnfr
.state
in [VnfRecordState
.INSTANTIATION_PENDING
, VnfRecordState
.INIT
]:
2146 self
._log
.debug("Received vnfr state for vnfr %s is %s; retrying",vnfr
.name
,vnfr
.state
)
2147 yield from asyncio
.sleep(2, loop
=self
._loop
)
2148 if vnfr
.state
== VnfRecordState
.ACTIVE
:
2149 self
._log
.debug("Received vnfr state for vnfr %s is %s ",vnfr
.name
,vnfr
.state
)
2152 self
._log
.debug("Received vnfr state for vnfr %s is %s; failing vnffg creation",vnfr
.name
,vnfr
.state
)
2153 self
._vnffgr
_state
= VnffgRecordState
.FAILED
2156 self
._log
.info("Waiting for 90 seconds for VMs to come up")
2157 yield from asyncio
.sleep(90, loop
=self
._loop
)
2158 self
._log
.info("Starting VNFFG orchestration")
2159 for vnffg
in self
._vnffgrs
.values():
2160 self
._log
.debug("Instantiating VNFFG: %s in NS %s", vnffg
, self
.id)
2161 yield from vnffg
.instantiate()
2164 def instantiate_scaling_instances(self
, config_xact
):
2165 """ Instantiate any default scaling instances in this Network Service """
2166 for group
in self
._scaling
_groups
.values():
2167 for i
in range(group
.min_instance_count
):
2168 self
._log
.debug("Instantiating %s default scaling instance %s", group
, i
)
2169 yield from self
.create_scale_group_instance(
2170 group
.name
, i
, config_xact
, is_default
=True
2173 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2174 if group_msg
.scaling_group_name_ref
!= group
.name
:
2177 for instance
in group_msg
.instance
:
2178 self
._log
.debug("Reloading %s scaling instance %s", group_msg
, instance
.id)
2179 yield from self
.create_scale_group_instance(
2180 group
.name
, instance
.id, config_xact
, is_default
=False
2183 def has_scaling_instances(self
):
2184 """ Return boolean indicating if the network service has default scaling groups """
2185 for group
in self
._scaling
_groups
.values():
2186 if group
.min_instance_count
> 0:
2189 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2190 if len(group_msg
.instance
) > 0:
2197 """ This function publishes this NSR """
2198 self
._nsr
_msg
= self
.create_msg()
2200 self
._log
.debug("Publishing the NSR with xpath %s and nsr %s",
2204 if self
._debug
_running
:
2205 self
._log
.debug("Publishing NSR in RUNNING state!")
2208 with self
._dts
.transaction() as xact
:
2209 yield from self
._nsm
.nsr_handler
.update(xact
, self
.nsr_xpath
, self
._nsr
_msg
)
2210 if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
:
2211 self
._debug
_running
= True
2214 def unpublish(self
, xact
):
2215 """ Unpublish this NSR object """
2216 self
._log
.debug("Unpublishing Network service id %s", self
.id)
2217 yield from self
._nsm
.nsr_handler
.delete(xact
, self
.nsr_xpath
)
2220 def nsr_xpath(self
):
2221 """ Returns the xpath associated with this NSR """
2222 return self
._project
.add_project((
2223 "D,/nsr:ns-instance-opdata" +
2224 "/nsr:nsr[nsr:ns-instance-config-ref = '{}']"
2228 def xpath_from_nsr(nsr
):
2229 """ Returns the xpath associated with this NSR op data"""
2230 return self
._project
.add_project((NetworkServiceRecord
.XPATH
+
2231 "[nsr:ns-instance-config-ref = '{}']").format(nsr
.id))
2234 def nsd_xpath(self
):
2235 """ Return NSD config xpath."""
2236 return self
._project
.add_project((
2237 "C,/project-nsd:nsd-catalog/project-nsd:nsd[project-nsd:id = '{}']"
2238 ).format(self
.nsd_id
))
2241 def instantiate(self
, config_xact
):
2242 """"Instantiates a NetworkServiceRecord.
2244 This function instantiates a Network service
2245 which involves the following steps,
2247 * Instantiate every VL in NSD by sending create VLR request to DTS.
2248 * Instantiate every VNF in NSD by sending create VNF reuqest to DTS.
2249 * Publish the NSR details to DTS
2252 nsr: The NSR configuration request containing nsr-id and nsd
2253 config_xact: The configuration transaction which initiated the instatiation
2256 NetworkServiceRecordError if the NSR creation fails
2262 self
._log
.debug("Instantiating NS - %s xact - %s", self
, config_xact
)
2264 # Move the state to INIITALIZING
2265 self
.set_state(NetworkServiceRecordState
.INIT
)
2267 event_descr
= "Instantiation Request Received NSR Id:%s" % self
.id
2268 self
.record_event("instantiating", event_descr
)
2271 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
2274 # Update ref count if nsd present in catalog
2275 self
._nsm
.get_nsd_ref(self
.nsd_id
)
2277 except NetworkServiceDescriptorError
:
2278 # This could be an NSD not in the nsd-catalog
2281 # Merge any config and initial config primitive values
2282 self
.config_store
.merge_nsd_config(self
.nsd_msg
)
2283 self
._log
.debug("Merged NSD: {}".format(self
.nsd_msg
.as_dict()))
2285 event_descr
= "Fetched NSD with descriptor id %s" % self
.nsd_id
2286 self
.record_event("nsd-fetched", event_descr
)
2288 if self
._nsd
is None:
2289 msg
= "Failed to fetch NSD with nsd-id [%s] for nsr-id %s"
2290 self
._log
.debug(msg
, self
.nsd_id
, self
.id)
2291 raise NetworkServiceRecordError(self
)
2293 self
._log
.debug("Got nsd result %s", self
._nsd
)
2295 # Substitute any input parameters
2296 self
.substitute_input_parameters(self
._nsd
, self
._nsr
_cfg
_msg
)
2299 yield from self
.create(config_xact
)
2301 # Publish the NSR to DTS
2302 yield from self
.publish()
2305 def do_instantiate():
2307 Instantiate network service
2309 self
._log
.debug("Instantiating VLs nsr id [%s] nsd id [%s]",
2310 self
.id, self
.nsd_id
)
2312 # instantiate the VLs
2313 event_descr
= ("Instantiating %s external VLs for NSR id %s" %
2314 (len(self
.nsd_msg
.vld
), self
.id))
2315 self
.record_event("begin-external-vls-instantiation", event_descr
)
2317 self
.set_state(NetworkServiceRecordState
.VL_INIT_PHASE
)
2319 yield from self
.instantiate_vls()
2321 # Publish the NSR to DTS
2322 yield from self
.publish()
2324 event_descr
= ("Finished instantiating %s external VLs for NSR id %s" %
2325 (len(self
.nsd_msg
.vld
), self
.id))
2326 self
.record_event("end-external-vls-instantiation", event_descr
)
2328 self
.set_state(NetworkServiceRecordState
.VNF_INIT_PHASE
)
2330 self
._log
.debug("Instantiating VNFs ...... nsr[%s], nsd[%s]",
2331 self
.id, self
.nsd_id
)
2333 # instantiate the VNFs
2334 event_descr
= ("Instantiating %s VNFS for NSR id %s" %
2335 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2337 self
.record_event("begin-vnf-instantiation", event_descr
)
2339 yield from self
.instantiate_vnfs(self
._vnfrs
.values())
2341 self
._log
.debug(" Finished instantiating %d VNFs for NSR id %s",
2342 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
2344 event_descr
= ("Finished instantiating %s VNFs for NSR id %s" %
2345 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2346 self
.record_event("end-vnf-instantiation", event_descr
)
2348 if len(self
.vnffgrs
) > 0:
2349 #self.set_state(NetworkServiceRecordState.VNFFG_INIT_PHASE)
2350 event_descr
= ("Instantiating %s VNFFGS for NSR id %s" %
2351 (len(self
.nsd_msg
.vnffgd
), self
.id))
2353 self
.record_event("begin-vnffg-instantiation", event_descr
)
2355 yield from self
.instantiate_vnffgs()
2357 event_descr
= ("Finished instantiating %s VNFFGDs for NSR id %s" %
2358 (len(self
.nsd_msg
.vnffgd
), self
.id))
2359 self
.record_event("end-vnffg-instantiation", event_descr
)
2361 if self
.has_scaling_instances():
2362 event_descr
= ("Instantiating %s Scaling Groups for NSR id %s" %
2363 (len(self
._scaling
_groups
), self
.id))
2365 self
.record_event("begin-scaling-group-instantiation", event_descr
)
2366 yield from self
.instantiate_scaling_instances(config_xact
)
2367 self
.record_event("end-scaling-group-instantiation", event_descr
)
2369 # Give the plugin a chance to deploy the network service now that all
2370 # virtual links and vnfs are instantiated
2371 yield from self
.nsm_plugin
.deploy(self
._nsr
_msg
)
2373 self
._log
.debug("Publishing NSR...... nsr[%s], nsd[%s]",
2374 self
.id, self
.nsd_id
)
2376 # Publish the NSR to DTS
2377 yield from self
.publish()
2379 self
._log
.debug("Published NSR...... nsr[%s], nsd[%s]",
2380 self
.id, self
.nsd_id
)
2382 def on_instantiate_done(fut
):
2383 # If the do_instantiate fails, then publish NSR with failed result
2384 if fut
.exception() is not None:
2385 self
._log
.error("NSR instantiation failed for NSR id %s: %s", self
.id, str(fut
.exception()))
2386 self
._loop
.create_task(self
.instantiation_failed(failed_reason
=str(fut
.exception())))
2388 instantiate_task
= self
._loop
.create_task(do_instantiate())
2389 instantiate_task
.add_done_callback(on_instantiate_done
)
2392 def set_config_status(self
, status
, status_details
=None):
2393 if self
.config_status
!= status
:
2394 self
._log
.debug("Updating NSR {} status for {} to {}".
2395 format(self
.name
, self
.config_status
, status
))
2396 self
._config
_status
= status
2397 self
._config
_status
_details
= status_details
2399 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2400 self
.record_event("config-failed", "NS configuration failed",
2401 evt_details
=self
._config
_status
_details
)
2403 yield from self
.publish()
2406 def is_active(self
):
2407 """ This NS is active """
2408 self
.set_state(NetworkServiceRecordState
.RUNNING
)
2412 # Publish the NSR to DTS
2413 self
._log
.debug("Network service %s is active ", self
.id)
2414 self
._is
_active
= True
2416 event_descr
= "NSR in running state for NSR id %s" % self
.id
2417 self
.record_event("ns-running", event_descr
)
2419 yield from self
.publish()
2422 def instantiation_failed(self
, failed_reason
=None):
2423 """ The NS instantiation failed"""
2424 self
._log
.error("Network service id:%s, name:%s instantiation failed",
2426 self
.set_state(NetworkServiceRecordState
.FAILED
)
2428 event_descr
= "Instantiation of NS %s failed" % self
.id
2429 self
.record_event("ns-failed", event_descr
, evt_details
=failed_reason
)
2431 # Publish the NSR to DTS
2432 yield from self
.publish()
2435 def terminate_vnfrs(self
, vnfrs
):
2436 """ Terminate VNFRS in this network service """
2437 self
._log
.debug("Terminating VNFs in network service %s", self
.id)
2439 yield from self
.nsm_plugin
.terminate_vnf(vnfr
)
2442 def terminate(self
):
2443 """ Terminate a NetworkServiceRecord."""
2444 def terminate_vnffgrs():
2445 """ Terminate VNFFGRS in this network service """
2446 self
._log
.debug("Terminating VNFFGRs in network service %s", self
.id)
2447 for vnffgr
in self
.vnffgrs
.values():
2448 yield from vnffgr
.terminate()
2450 def terminate_vlrs():
2451 """ Terminate VLRs in this netork service """
2452 self
._log
.debug("Terminating VLs in network service %s", self
.id)
2453 for vlr
in self
.vlrs
:
2454 yield from self
.nsm_plugin
.terminate_vl(vlr
)
2455 vlr
.state
= VlRecordState
.TERMINATED
2456 if vlr
.id in self
.vlr_uptime_tasks
:
2457 self
.vlr_uptime_tasks
[vlr
.id].cancel()
2459 self
._log
.debug("Terminating network service id %s", self
.id)
2461 # Move the state to TERMINATE
2462 self
.set_state(NetworkServiceRecordState
.TERMINATE
)
2463 event_descr
= "Terminate being processed for NS Id:%s" % self
.id
2464 self
.record_event("terminate", event_descr
)
2466 # Move the state to VNF_TERMINATE_PHASE
2467 self
._log
.debug("Terminating VNFFGs in NS ID: %s", self
.id)
2468 self
.set_state(NetworkServiceRecordState
.VNFFG_TERMINATE_PHASE
)
2469 event_descr
= "Terminating VNFFGS in NS Id:%s" % self
.id
2470 self
.record_event("terminating-vnffgss", event_descr
)
2471 yield from terminate_vnffgrs()
2473 # Move the state to VNF_TERMINATE_PHASE
2474 self
.set_state(NetworkServiceRecordState
.VNF_TERMINATE_PHASE
)
2475 event_descr
= "Terminating VNFS in NS Id:%s" % self
.id
2476 self
.record_event("terminating-vnfs", event_descr
)
2477 yield from self
.terminate_vnfrs(self
.vnfrs
.values())
2479 # Move the state to VL_TERMINATE_PHASE
2480 self
.set_state(NetworkServiceRecordState
.VL_TERMINATE_PHASE
)
2481 event_descr
= "Terminating VLs in NS Id:%s" % self
.id
2482 self
.record_event("terminating-vls", event_descr
)
2483 yield from terminate_vlrs()
2485 yield from self
.nsm_plugin
.terminate_ns(self
)
2487 # Move the state to TERMINATED
2488 self
.set_state(NetworkServiceRecordState
.TERMINATED
)
2489 event_descr
= "Terminated NS Id:%s" % self
.id
2490 self
.record_event("terminated", event_descr
)
2493 """"Enable a NetworkServiceRecord."""
2497 """"Disable a NetworkServiceRecord."""
2500 def map_config_status(self
):
2501 self
._log
.debug("Config status for ns {} is {}".
2502 format(self
.name
, self
._config
_status
))
2503 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURING
:
2504 return 'configuring'
2505 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2509 def vl_phase_completed(self
):
2510 """ Are VLs created in this NS?"""
2511 return self
._vl
_phase
_completed
2513 def vnf_phase_completed(self
):
2514 """ Are VLs created in this NS?"""
2515 return self
._vnf
_phase
_completed
2517 def create_msg(self
):
2518 """ The network serice record as a message """
2519 nsr_dict
= {"ns_instance_config_ref": self
.id}
2520 nsr
= RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr
.from_dict(nsr_dict
)
2521 #nsr.cloud_account = self.cloud_account_name
2522 nsr
.sdn_account
= self
._sdn
_account
_name
2523 nsr
.name_ref
= self
.name
2524 nsr
.nsd_ref
= self
.nsd_id
2525 nsr
.nsd_name_ref
= self
.nsd_msg
.name
2526 nsr
.operational_events
= self
._op
_status
.msg
2527 nsr
.operational_status
= self
._op
_status
.yang_str()
2528 nsr
.config_status
= self
.map_config_status()
2529 nsr
.config_status_details
= self
._config
_status
_details
2530 nsr
.create_time
= self
._create
_time
2531 nsr
.uptime
= int(time
.time()) - self
._create
_time
2533 for cfg_prim
in self
.nsd_msg
.service_primitive
:
2534 cfg_prim
= NsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr_ServicePrimitive
.from_dict(
2536 nsr
.service_primitive
.append(cfg_prim
)
2538 for init_cfg
in self
.nsd_msg
.initial_config_primitive
:
2539 prim
= NsrYang
.NsrInitialConfigPrimitive
.from_dict(
2541 nsr
.initial_config_primitive
.append(prim
)
2543 if self
.vl_phase_completed():
2544 for vlr
in self
.vlrs
:
2545 nsr
.vlr
.append(vlr
.create_nsr_vlr_msg(self
.vnfrs
.values()))
2547 if self
.vnf_phase_completed():
2548 for vnfr_id
in self
.vnfrs
:
2549 nsr
.constituent_vnfr_ref
.append(self
.vnfrs
[vnfr_id
].const_vnfr_msg
)
2550 for vnffgr
in self
.vnffgrs
.values():
2551 nsr
.vnffgr
.append(vnffgr
.fetch_vnffgr())
2552 for scaling_group
in self
._scaling
_groups
.values():
2553 nsr
.scaling_group_record
.append(scaling_group
.create_record_msg())
2557 def all_vnfs_active(self
):
2558 """ Are all VNFS in this NS active? """
2559 for _
, vnfr
in self
.vnfrs
.items():
2560 if vnfr
.active
is not True:
2565 def update_state(self
):
2566 """ Re-evaluate this NS's state """
2567 curr_state
= self
._op
_status
.state
2569 if curr_state
== NetworkServiceRecordState
.TERMINATED
:
2570 self
._log
.debug("NS (%s) in terminated state, not updating state", self
.id)
2573 new_state
= NetworkServiceRecordState
.RUNNING
2574 self
._log
.info("Received update_state for nsr: %s, curr-state: %s",
2575 self
.id, curr_state
)
2577 # Check all the VNFRs are present
2578 for _
, vnfr
in self
.vnfrs
.items():
2579 if vnfr
.state
in [VnfRecordState
.ACTIVE
, VnfRecordState
.TERMINATED
]:
2581 elif vnfr
.state
== VnfRecordState
.FAILED
:
2582 if vnfr
._prev
_state
!= vnfr
.state
:
2583 event_descr
= "Instantiation of VNF %s failed" % vnfr
.id
2584 event_error_details
= vnfr
.state_failed_reason
2585 self
.record_event("vnf-failed", event_descr
, evt_details
=event_error_details
)
2586 vnfr
.set_state(VnfRecordState
.FAILED
)
2588 self
._log
.info("VNF state did not change, curr=%s, prev=%s",
2589 vnfr
.state
, vnfr
._prev
_state
)
2590 new_state
= NetworkServiceRecordState
.FAILED
2593 self
._log
.info("VNF %s in NSR %s is still not active; current state is: %s",
2594 vnfr
.id, self
.id, vnfr
.state
)
2595 new_state
= curr_state
2597 # If new state is RUNNING; check all VLs
2598 if new_state
== NetworkServiceRecordState
.RUNNING
:
2599 for vl
in self
.vlrs
:
2601 if vl
.state
in [VlRecordState
.ACTIVE
, VlRecordState
.TERMINATED
]:
2603 elif vl
.state
== VlRecordState
.FAILED
:
2604 if vl
.prev_state
!= vl
.state
:
2605 event_descr
= "Instantiation of VL %s failed" % vl
.id
2606 event_error_details
= vl
.state_failed_reason
2607 self
.record_event("vl-failed", event_descr
, evt_details
=event_error_details
)
2608 vl
.prev_state
= vl
.state
2610 self
._log
.debug("VL %s already in failed state")
2612 if vl
.state
in [VlRecordState
.INSTANTIATION_PENDING
, VlRecordState
.INIT
]:
2613 new_state
= NetworkServiceRecordState
.VL_INSTANTIATE
2616 if vl
.state
in [VlRecordState
.TERMINATE_PENDING
]:
2617 new_state
= NetworkServiceRecordState
.VL_TERMINATE
2620 # If new state is RUNNING; check VNFFGRs are also active
2621 if new_state
== NetworkServiceRecordState
.RUNNING
:
2622 for _
, vnffgr
in self
.vnffgrs
.items():
2623 self
._log
.info("Checking vnffgr state for nsr %s is: %s",
2624 self
.id, vnffgr
.state
)
2625 if vnffgr
.state
== VnffgRecordState
.ACTIVE
:
2627 elif vnffgr
.state
== VnffgRecordState
.FAILED
:
2628 event_descr
= "Instantiation of VNFFGR %s failed" % vnffgr
.id
2629 self
.record_event("vnffg-failed", event_descr
)
2630 new_state
= NetworkServiceRecordState
.FAILED
2633 self
._log
.info("VNFFGR %s in NSR %s is still not active; current state is: %s",
2634 vnffgr
.id, self
.id, vnffgr
.state
)
2635 new_state
= curr_state
2637 # Update all the scaling group instance operational status to
2638 # reflect the state of all VNFR within that instance
2639 yield from self
._update
_scale
_group
_instances
_status
()
2641 for _
, group
in self
._scaling
_groups
.items():
2642 if group
.state
== scale_group
.ScaleGroupState
.SCALING_OUT
:
2643 new_state
= NetworkServiceRecordState
.SCALING_OUT
2645 elif group
.state
== scale_group
.ScaleGroupState
.SCALING_IN
:
2646 new_state
= NetworkServiceRecordState
.SCALING_IN
2649 if new_state
!= curr_state
:
2650 self
._log
.debug("Changing state of Network service %s from %s to %s",
2651 self
.id, curr_state
, new_state
)
2652 if new_state
== NetworkServiceRecordState
.RUNNING
:
2653 yield from self
.is_active()
2654 elif new_state
== NetworkServiceRecordState
.FAILED
:
2655 # If the NS is already active and we entered scaling_in, scaling_out,
2656 # do not mark the NS as failing if scaling operation failed.
2657 if curr_state
in [NetworkServiceRecordState
.SCALING_OUT
,
2658 NetworkServiceRecordState
.SCALING_IN
] and self
._is
_active
:
2659 new_state
= NetworkServiceRecordState
.RUNNING
2660 self
.set_state(new_state
)
2662 yield from self
.instantiation_failed()
2664 self
.set_state(new_state
)
2666 yield from self
.publish()
2669 class InputParameterSubstitution(object):
2671 This class is responsible for substituting input parameters into an NSD.
2674 def __init__(self
, log
, project
):
2675 """Create an instance of InputParameterSubstitution
2678 log - a logger for this object to use
2682 self
.project
= project
2684 def __call__(self
, nsd
, nsr_config
):
2685 """Substitutes input parameters from the NSR config into the NSD
2687 This call modifies the provided NSD with the input parameters that are
2688 contained in the NSR config.
2691 nsd - a GI NSD object
2692 nsr_config - a GI NSR config object
2695 if nsd
is None or nsr_config
is None:
2698 # Create a lookup of the xpath elements that this descriptor allows
2700 optional_input_parameters
= set()
2701 for input_parameter
in nsd
.input_parameter_xpath
:
2702 optional_input_parameters
.add(self
.project
.add_project(input_parameter
.xpath
))
2704 # Apply the input parameters to the descriptor
2705 if nsr_config
.input_parameter
:
2706 for param
in nsr_config
.input_parameter
:
2707 if param
.xpath
not in optional_input_parameters
:
2708 msg
= "tried to set an invalid input parameter ({})"
2709 self
.log
.error(msg
.format(param
.xpath
))
2713 "input-parameter:{} = {}".format(
2720 xpath
.setxattr(nsd
, param
.xpath
, param
.value
)
2722 except Exception as e
:
2723 self
.log
.exception(e
)
2726 class NetworkServiceDescriptor(object):
2728 Network service descriptor class
2731 def __init__(self
, dts
, log
, loop
, nsd
, nsm
):
2743 """ Returns nsd id """
2748 """ Returns name of nsd """
2749 return self
._nsd
.name
2752 def ref_count(self
):
2753 """ Returns reference count"""
2754 return self
._ref
_count
2757 """ Returns whether nsd is in use or not """
2758 return True if self
.ref_count
> 0 else False
2761 """ Take a reference on this object """
2762 self
._ref
_count
+= 1
2765 """ Release reference on this object """
2766 if self
.ref_count
< 1:
2767 msg
= ("Unref on a NSD object - nsd id %s, ref_count = %s" %
2768 (self
.id, self
.ref_count
))
2769 self
._log
.critical(msg
)
2770 raise NetworkServiceDescriptorError(msg
)
2771 self
._ref
_count
-= 1
2775 """ Return the message associated with this NetworkServiceDescriptor"""
2779 def path_for_id(nsd_id
):
2780 """ Return path for the passed nsd_id"""
2781 return self
._nsm
._project
.add_project(
2782 "C,/project-nsd:nsd-catalog/project-nsd:nsd[project-nsd:id = '{}'".
2786 """ Return the message associated with this NetworkServiceDescriptor"""
2787 return NetworkServiceDescriptor
.path_for_id(self
.id)
2789 def update(self
, nsd
):
2790 """ Update the NSD descriptor """
2794 class NsdDtsHandler(object):
2795 """ The network service descriptor DTS handler """
2796 XPATH
= "C,/project-nsd:nsd-catalog/project-nsd:nsd"
2798 def __init__(self
, dts
, log
, loop
, nsm
):
2805 self
._project
= nsm
._project
2809 """ Return registration handle """
2814 """ Register for Nsd create/update/delete/read requests from dts """
2817 self
._log
.warning("DTS handler already registered for project {}".
2818 format(self
._project
.name
))
2821 def on_apply(dts
, acg
, xact
, action
, scratch
):
2822 """Apply the configuration"""
2823 is_recovery
= xact
.xact
is None and action
== rwdts
.AppconfAction
.INSTALL
2824 self
._log
.debug("Got nsd apply cfg (xact:%s) (action:%s)",
2826 # Create/Update an NSD record
2827 for cfg
in self
._regh
.get_xact_elements(xact
):
2828 # Only interested in those NSD cfgs whose ID was received in prepare callback
2829 if cfg
.id in scratch
.get('nsds', []) or is_recovery
:
2830 self
._nsm
.update_nsd(cfg
)
2832 scratch
.pop('nsds', None)
2834 return RwTypes
.RwStatus
.SUCCESS
2837 def delete_nsd_libs(nsd_id
):
2838 """ Remove any files uploaded with NSD and stored under $RIFT_ARTIFACTS/libs/<id> """
2840 rift_artifacts_dir
= os
.environ
['RIFT_ARTIFACTS']
2841 nsd_dir
= os
.path
.join(rift_artifacts_dir
, 'launchpad/libs', nsd_id
)
2843 if os
.path
.exists (nsd_dir
):
2844 shutil
.rmtree(nsd_dir
, ignore_errors
=True)
2845 except Exception as e
:
2846 self
._log
.error("Exception in cleaning up NSD libs {}: {}".
2848 self
._log
.exception(e
)
2851 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2852 """ Prepare callback from DTS for NSD config """
2854 self
._log
.info("Got nsd prepare - config received nsd id %s, msg %s",
2857 fref
= ProtobufC
.FieldReference
.alloc()
2858 fref
.goto_whole_message(msg
.to_pbcm())
2860 if fref
.is_field_deleted():
2861 # Delete an NSD record
2862 self
._log
.debug("Deleting NSD with id %s", msg
.id)
2863 if self
._nsm
.nsd_in_use(msg
.id):
2864 self
._log
.debug("Cannot delete NSD in use - %s", msg
.id)
2865 err
= "Cannot delete an NSD in use - %s" % msg
.id
2866 raise NetworkServiceDescriptorRefCountExists(err
)
2868 yield from delete_nsd_libs(msg
.id)
2869 self
._nsm
.delete_nsd(msg
.id)
2871 # Add this NSD to scratch to create/update in apply callback
2872 nsds
= scratch
.setdefault('nsds', [])
2874 # acg._scratch['nsds'].append(msg.id)
2876 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2879 "Registering for NSD config using xpath: %s",
2880 NsdDtsHandler
.XPATH
,
2883 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2884 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2885 # Need a list in scratch to store NSDs to create/update later
2886 # acg._scratch['nsds'] = list()
2887 self
._regh
= acg
.register(
2888 xpath
=self
._project
.add_project(NsdDtsHandler
.XPATH
),
2889 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
2890 on_prepare
=on_prepare
)
2892 def deregister(self
):
2893 self
._log
.debug("De-register NSD handler for project {}".
2894 format(self
._project
.name
))
2896 self
._regh
.deregister()
2900 class VnfdDtsHandler(object):
2901 """ DTS handler for VNFD config changes """
2902 XPATH
= "C,/project-vnfd:vnfd-catalog/project-vnfd:vnfd"
2904 def __init__(self
, dts
, log
, loop
, nsm
):
2910 self
._project
= nsm
._project
2914 """ DTS registration handle """
2919 """ Register for VNFD configuration"""
2922 self
._log
.warning("DTS handler already registered for project {}".
2923 format(self
._project
.name
))
2927 def on_apply(dts
, acg
, xact
, action
, scratch
):
2928 """Apply the configuration"""
2929 self
._log
.debug("Got NSM VNFD apply (xact: %s) (action: %s)(scr: %s)",
2930 xact
, action
, scratch
)
2932 # Create/Update a VNFD record
2933 for cfg
in self
._regh
.get_xact_elements(xact
):
2934 # Only interested in those VNFD cfgs whose ID was received in prepare callback
2935 if cfg
.id in scratch
.get('vnfds', []):
2936 self
._nsm
.update_vnfd(cfg
)
2938 for cfg
in self
._regh
.elements
:
2939 if cfg
.id in scratch
.get('deleted_vnfds', []):
2940 yield from self
._nsm
.delete_vnfd(cfg
.id)
2942 scratch
.pop('vnfds', None)
2943 scratch
.pop('deleted_vnfds', None)
2946 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2947 """ on prepare callback """
2948 self
._log
.debug("Got on prepare for VNFD (path: %s) (action: %s) (msg: %s)",
2949 ks_path
.to_xpath(RwNsmYang
.get_schema()), xact_info
.query_action
, msg
)
2951 fref
= ProtobufC
.FieldReference
.alloc()
2952 fref
.goto_whole_message(msg
.to_pbcm())
2954 # Handle deletes in prepare_callback, but adds/updates in apply_callback
2955 if fref
.is_field_deleted():
2956 self
._log
.debug("Adding msg to deleted field")
2957 deleted_vnfds
= scratch
.setdefault('deleted_vnfds', [])
2958 deleted_vnfds
.append(msg
.id)
2960 # Add this VNFD to scratch to create/update in apply callback
2961 vnfds
= scratch
.setdefault('vnfds', [])
2962 vnfds
.append(msg
.id)
2964 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2966 xpath
= self
._project
.add_project(VnfdDtsHandler
.XPATH
)
2968 "Registering for VNFD config using xpath {} for project {}"
2969 .format(xpath
, self
._project
))
2970 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2971 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2972 # Need a list in scratch to store VNFDs to create/update later
2973 # acg._scratch['vnfds'] = list()
2974 # acg._scratch['deleted_vnfds'] = list()
2975 self
._regh
= acg
.register(
2977 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY
,
2978 on_prepare
=on_prepare
)
2980 def deregister(self
):
2981 self
._log
.debug("De-register VNFD handler for project {}".
2982 format(self
._project
.name
))
2984 self
._regh
.deregister()
2988 class NsrRpcDtsHandler(object):
2989 """ The network service instantiation RPC DTS handler """
2990 EXEC_NSR_CONF_XPATH
= "I,/nsr:start-network-service"
2991 EXEC_NSR_CONF_O_XPATH
= "O,/nsr:start-network-service"
2992 NETCONF_IP_ADDRESS
= "127.0.0.1"
2994 RESTCONF_PORT
= 8888
2995 NETCONF_USER
= "admin"
2996 NETCONF_PW
= "admin"
2997 REST_BASE_V2_URL
= 'https://{}:{}/v2/api/'.format("127.0.0.1",8888)
2999 def __init__(self
, dts
, log
, loop
, nsm
):
3006 self
._ns
_regh
= None
3008 self
._manager
= None
3009 self
._nsr
_config
_url
= NsrRpcDtsHandler
.REST_BASE_V2_URL
+ \
3010 'config/project/{}/ns-instance-config'. \
3011 format(self
._nsm
._project
.name
)
3013 self
._model
= RwYang
.Model
.create_libncx()
3014 self
._model
.load_schema_ypbc(RwNsrYang
.get_schema())
3018 """ Return the NS manager instance """
3022 def wrap_netconf_config_xml(xml
):
3023 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
3027 def _connect(self
, timeout_secs
=240):
3029 start_time
= time
.time()
3030 while (time
.time() - start_time
) < timeout_secs
:
3033 self
._log
.debug("Attemping NsmTasklet netconf connection.")
3035 manager
= yield from ncclient
.asyncio_manager
.asyncio_connect(
3037 host
=NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
,
3038 port
=NsrRpcDtsHandler
.NETCONF_PORT
,
3039 username
=NsrRpcDtsHandler
.NETCONF_USER
,
3040 password
=NsrRpcDtsHandler
.NETCONF_PW
,
3042 look_for_keys
=False,
3043 hostkey_verify
=False,
3048 except ncclient
.transport
.errors
.SSHError
as e
:
3049 self
._log
.warning("Netconf connection to launchpad %s failed: %s",
3050 NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
, str(e
))
3052 yield from asyncio
.sleep(5, loop
=self
._loop
)
3054 raise NsrInstantiationFailed("Failed to connect to Launchpad within %s seconds" %
3057 def _apply_ns_instance_config(self
,payload_dict
):
3058 #self._log.debug("At apply NS instance config with payload %s",payload_dict)
3059 req_hdr
= {'accept':'application/vnd.yang.data+json',
3060 'content-type':'application/vnd.yang.data+json'}
3061 response
=requests
.post(self
._nsr
_config
_url
, headers
=req_hdr
,
3062 auth
=('admin', 'admin'),data
=payload_dict
,verify
=False)
3067 """ Register for NS monitoring read from dts """
3069 self
._log
.warning("RPC already registered for project {}".
3070 format(self
._project
.name
))
3074 def on_ns_config_prepare(xact_info
, action
, ks_path
, msg
):
3075 """ prepare callback from dts start-network-service"""
3076 assert action
== rwdts
.QueryAction
.RPC
3079 if not self
._nsm
._project
.rpc_check(msg
, xact_info
=xact_info
):
3082 rpc_op
= NsrYang
.YangOutput_Nsr_StartNetworkService
.from_dict({
3083 "nsr_id":str(uuid
.uuid4()),
3084 "project_name": msg
.prject_name
,
3087 if not ('name' in rpc_ip
and 'nsd_ref' in rpc_ip
and
3088 ('cloud_account' in rpc_ip
or 'om_datacenter' in rpc_ip
)):
3089 self
._log
.error("Mandatory parameters name or nsd_ref or " +
3090 "cloud account not found in start-network-service {}".
3094 self
._log
.debug("start-network-service RPC input: {}".format(rpc_ip
))
3097 # Add used value to the pool
3098 self
._log
.debug("RPC output: {}".format(rpc_op
))
3100 nsd_copy
= self
.nsm
.get_nsd(rpc_ip
.nsd_ref
)
3102 #if not self._manager:
3103 # self._manager = yield from self._connect()
3105 self
._log
.debug("Configuring ns-instance-config with name %s nsd-ref: %s",
3106 rpc_ip
.name
, rpc_ip
.nsd_ref
)
3108 ns_instance_config_dict
= {"id":rpc_op
.nsr_id
, "admin_status":"ENABLED"}
3109 ns_instance_config_copy_dict
= {k
:v
for k
, v
in rpc_ip
.as_dict().items()
3110 if k
in RwNsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr().fields
}
3111 ns_instance_config_dict
.update(ns_instance_config_copy_dict
)
3113 ns_instance_config
= RwNsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr
.from_dict(ns_instance_config_dict
)
3114 ns_instance_config
.nsd
= NsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr_Nsd()
3115 ns_instance_config
.nsd
.from_dict(nsd_copy
.msg
.as_dict())
3117 payload_dict
= ns_instance_config
.to_json(self
._model
)
3118 #xml = ns_instance_config.to_xml_v2(self._model)
3119 #netconf_xml = self.wrap_netconf_config_xml(xml)
3121 #self._log.debug("Sending configure ns-instance-config xml to %s: %s",
3122 # netconf_xml, NsrRpcDtsHandler.NETCONF_IP_ADDRESS)
3123 self
._log
.debug("Sending configure ns-instance-config json to %s: %s",
3124 self
._nsr
_config
_url
,ns_instance_config
)
3126 #response = yield from self._manager.edit_config(
3128 # config=netconf_xml,
3130 response
= yield from self
._loop
.run_in_executor(
3132 self
._apply
_ns
_instance
_config
,
3135 response
.raise_for_status()
3136 self
._log
.debug("Received edit config response: %s", response
.json())
3138 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
,
3139 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
,
3141 except Exception as e
:
3142 self
._log
.error("Exception processing the "
3143 "start-network-service: {}".format(e
))
3144 self
._log
.exception(e
)
3145 xact_info
.respond_xpath(rwdts
.XactRspCode
.NACK
,
3146 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
)
3149 hdl_ns
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_ns_config_prepare
,)
3151 with self
._dts
.group_create() as group
:
3152 self
._ns
_regh
= group
.register(xpath
=NsrRpcDtsHandler
.EXEC_NSR_CONF_XPATH
,
3154 flags
=rwdts
.Flag
.PUBLISHER
,
3157 def deregister(self
):
3158 self
._log
.debug("De-register NSR RPC for project {}".
3159 format(self
._nsm
._project
.name
))
3161 self
._ns
_regh
.deregister()
3162 self
._ns
_regh
= None
3165 class NsrDtsHandler(object):
3166 """ The network service DTS handler """
3167 NSR_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr"
3168 SCALE_INSTANCE_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr/nsr:scaling-group/nsr:instance"
3169 KEY_PAIR_XPATH
= "C,/nsr:key-pair"
3171 def __init__(self
, dts
, log
, loop
, nsm
):
3176 self
._project
= self
._nsm
._project
3178 self
._nsr
_regh
= None
3179 self
._scale
_regh
= None
3180 self
._key
_pair
_regh
= None
3184 """ Return the NS manager instance """
3189 """ Register for Nsr create/update/delete/read requests from dts """
3192 self
._log
.warning("DTS handler already registered for project {}".
3193 format(self
._project
.name
))
3196 def nsr_id_from_keyspec(ks
):
3197 nsr_path_entry
= NsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr
.schema().keyspec_to_entry(ks
)
3198 nsr_id
= nsr_path_entry
.key00
.id
3201 def group_name_from_keyspec(ks
):
3202 group_path_entry
= NsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup
.schema().keyspec_to_entry(ks
)
3203 group_name
= group_path_entry
.key00
.scaling_group_name_ref
3206 def is_instance_in_reg_elements(nsr_id
, group_name
, instance_id
):
3207 """ Return boolean indicating if scaling group instance was already commited previously.
3209 By looking at the existing elements in this registration handle (elements not part
3210 of this current xact), we can tell if the instance was configured previously without
3211 keeping any application state.
3213 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3214 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3215 elem_group_name
= group_name_from_keyspec(keyspec
)
3217 if elem_nsr_id
!= nsr_id
or group_name
!= elem_group_name
:
3220 if instance_cfg
.id == instance_id
:
3225 def get_scale_group_instance_delta(nsr_id
, group_name
, xact
):
3226 delta
= {"added": [], "deleted": []}
3227 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3228 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3229 if elem_nsr_id
!= nsr_id
:
3232 elem_group_name
= group_name_from_keyspec(keyspec
)
3233 if elem_group_name
!= group_name
:
3236 delta
["added"].append(instance_cfg
.id)
3238 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(include_keyspec
=True):
3239 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3240 if elem_nsr_id
!= nsr_id
:
3243 elem_group_name
= group_name_from_keyspec(keyspec
)
3244 if elem_group_name
!= group_name
:
3247 if instance_cfg
.id in delta
["added"]:
3248 delta
["added"].remove(instance_cfg
.id)
3250 delta
["deleted"].append(instance_cfg
.id)
3255 def update_nsr_nsd(nsr_id
, xact
, scratch
):
3258 def get_nsr_vl_delta(nsr_id
, xact
, scratch
):
3259 delta
= {"added": [], "deleted": []}
3260 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3261 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3262 if elem_nsr_id
!= nsr_id
:
3265 if 'vld' in instance_cfg
.nsd
:
3266 for vld
in instance_cfg
.nsd
.vld
:
3267 delta
["added"].append(vld
)
3269 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3270 self
._log
.debug("NSR update: %s", instance_cfg
)
3271 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3272 if elem_nsr_id
!= nsr_id
:
3275 if 'vld' in instance_cfg
.nsd
:
3276 for vld
in instance_cfg
.nsd
.vld
:
3277 if vld
in delta
["added"]:
3278 delta
["added"].remove(vld
)
3280 delta
["deleted"].append(vld
)
3284 vl_delta
= yield from get_nsr_vl_delta(nsr_id
, xact
, scratch
)
3285 self
._log
.debug("Got NSR:%s VL instance delta: %s", nsr_id
, vl_delta
)
3287 for vld
in vl_delta
["added"]:
3288 yield from self
._nsm
.nsr_instantiate_vl(nsr_id
, vld
)
3290 for vld
in vl_delta
["deleted"]:
3291 yield from self
._nsm
.nsr_terminate_vl(nsr_id
, vld
)
3293 def get_nsr_key_pairs(dts_member_reg
, xact
):
3295 for instance_cfg
, keyspec
in dts_member_reg
.get_xact_elements(xact
, include_keyspec
=True):
3296 self
._log
.debug("Key pair received is {} KS: {}".format(instance_cfg
, keyspec
))
3297 xpath
= keyspec
.to_xpath(RwNsrYang
.get_schema())
3298 key_pairs
[instance_cfg
.name
] = instance_cfg
3301 def on_apply(dts
, acg
, xact
, action
, scratch
):
3302 """Apply the configuration"""
3303 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3304 xact
, action
, scratch
)
3306 def handle_create_nsr(msg
, key_pairs
=None, restart_mode
=False):
3307 # Handle create nsr requests """
3308 # Do some validations
3309 if not msg
.has_field("nsd"):
3310 err
= "NSD not provided"
3311 self
._log
.error(err
)
3312 raise NetworkServiceRecordError(err
)
3314 self
._log
.debug("Creating NetworkServiceRecord %s from nsr config %s",
3315 msg
.id, msg
.as_dict())
3316 nsr
= self
.nsm
.create_nsr(msg
, key_pairs
=key_pairs
, restart_mode
=restart_mode
)
3319 def handle_delete_nsr(msg
):
3321 def delete_instantiation(ns_id
):
3322 """ Delete instantiation """
3323 with self
._dts
.transaction() as xact
:
3324 yield from self
._nsm
.terminate_ns(ns_id
, xact
)
3326 # Handle delete NSR requests
3327 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3328 # Terminate the NSR instance
3329 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3331 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3332 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3333 nsr
.record_event("terminate-rcvd", event_descr
)
3335 self
._loop
.create_task(delete_instantiation(msg
.id))
3338 def begin_instantiation(nsr
):
3339 # Begin instantiation
3340 self
._log
.info("Beginning NS instantiation: %s", nsr
.id)
3342 yield from self
._nsm
.instantiate_ns(nsr
.id, xact
)
3343 except Exception as e
:
3344 self
._log
.exception("NS instantiation: {}".format(e
))
3347 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3348 xact
, action
, scratch
)
3350 if action
== rwdts
.AppconfAction
.INSTALL
and xact
.id is None:
3352 for element
in self
._key
_pair
_regh
.elements
:
3353 key_pairs
.append(element
)
3354 for element
in self
._nsr
_regh
.elements
:
3355 nsr
= handle_create_nsr(element
, key_pairs
, restart_mode
=True)
3356 self
._loop
.create_task(begin_instantiation(nsr
))
3359 (added_msgs
, deleted_msgs
, updated_msgs
) = get_add_delete_update_cfgs(self
._nsr
_regh
,
3362 self
._log
.debug("Added: %s, Deleted: %s, Updated: %s", added_msgs
,
3363 deleted_msgs
, updated_msgs
)
3365 for msg
in added_msgs
:
3366 if msg
.id not in self
._nsm
.nsrs
:
3367 self
._log
.info("Create NSR received in on_apply to instantiate NS:%s", msg
.id)
3368 key_pairs
= get_nsr_key_pairs(self
._key
_pair
_regh
, xact
)
3369 nsr
= handle_create_nsr(msg
,key_pairs
)
3370 self
._loop
.create_task(begin_instantiation(nsr
))
3372 for msg
in deleted_msgs
:
3373 self
._log
.info("Delete NSR received in on_apply to terminate NS:%s", msg
.id)
3375 handle_delete_nsr(msg
)
3377 self
._log
.exception("Failed to terminate NS:%s", msg
.id)
3379 for msg
in updated_msgs
:
3380 self
._log
.info("Update NSR received in on_apply: %s", msg
)
3382 self
._nsm
.nsr_update_cfg(msg
.id, msg
)
3385 self
._loop
.create_task(update_nsr_nsd(msg
.id, xact
, scratch
))
3387 for group
in msg
.scaling_group
:
3388 instance_delta
= get_scale_group_instance_delta(msg
.id, group
.scaling_group_name_ref
, xact
)
3389 self
._log
.debug("Got NSR:%s scale group instance delta: %s", msg
.id, instance_delta
)
3391 for instance_id
in instance_delta
["added"]:
3392 self
._nsm
.scale_nsr_out(msg
.id, group
.scaling_group_name_ref
, instance_id
, xact
)
3394 for instance_id
in instance_delta
["deleted"]:
3395 self
._nsm
.scale_nsr_in(msg
.id, group
.scaling_group_name_ref
, instance_id
)
3398 return RwTypes
.RwStatus
.SUCCESS
3401 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
3402 """ Prepare calllback from DTS for NSR """
3404 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3405 action
= xact_info
.query_action
3407 "Got Nsr prepare callback (xact: %s) (action: %s) (info: %s), %s:%s)",
3408 xact
, action
, xact_info
, xpath
, msg
3412 def delete_instantiation(ns_id
):
3413 """ Delete instantiation """
3414 yield from self
._nsm
.terminate_ns(ns_id
, None)
3416 def handle_delete_nsr():
3417 """ Handle delete NSR requests """
3418 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3419 # Terminate the NSR instance
3420 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3422 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3423 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3424 nsr
.record_event("terminate-rcvd", event_descr
)
3426 self
._loop
.create_task(delete_instantiation(msg
.id))
3428 fref
= ProtobufC
.FieldReference
.alloc()
3429 fref
.goto_whole_message(msg
.to_pbcm())
3431 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
, rwdts
.QueryAction
.DELETE
]:
3432 # if this is an NSR create
3433 if action
!= rwdts
.QueryAction
.DELETE
and msg
.id not in self
._nsm
.nsrs
:
3434 # Ensure the Cloud account/datacenter has been specified
3435 if not msg
.has_field("cloud_account") and not msg
.has_field("om_datacenter"):
3436 raise NsrInstantiationFailed("Cloud account or datacenter not specified in NSR")
3438 # Check if nsd is specified
3439 if not msg
.has_field("nsd"):
3440 raise NsrInstantiationFailed("NSD not specified in NSR")
3443 nsr
= self
._nsm
.nsrs
[msg
.id]
3445 if msg
.has_field("nsd"):
3446 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3447 raise NsrVlUpdateError("Unable to update VL when NSR not in running state")
3448 if 'vld' not in msg
.nsd
or len(msg
.nsd
.vld
) == 0:
3449 raise NsrVlUpdateError("NS config NSD should have atleast 1 VLD defined")
3451 if msg
.has_field("scaling_group"):
3452 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3453 raise ScalingOperationError("Unable to perform scaling action when NS is not in running state")
3455 if len(msg
.scaling_group
) > 1:
3456 raise ScalingOperationError("Only a single scaling group can be configured at a time")
3458 for group_msg
in msg
.scaling_group
:
3459 num_new_group_instances
= len(group_msg
.instance
)
3460 if num_new_group_instances
> 1:
3461 raise ScalingOperationError("Only a single scaling instance can be modified at a time")
3463 elif num_new_group_instances
== 1:
3464 scale_group
= nsr
.scaling_groups
[group_msg
.scaling_group_name_ref
]
3465 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
]:
3466 if len(scale_group
.instances
) == scale_group
.max_instance_count
:
3467 raise ScalingOperationError("Max instances for %s reached" % scale_group
)
3469 acg
.handle
.prepare_complete_ok(xact_info
.handle
)
3472 xpath
= self
._project
.add_project(NsrDtsHandler
.NSR_XPATH
)
3473 self
._log
.debug("Registering for NSR config using xpath: {}".
3476 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
3477 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
3478 self
._nsr
_regh
= acg
.register(
3480 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3481 on_prepare
=on_prepare
3484 self
._scale
_regh
= acg
.register(
3485 xpath
=self
._project
.add_project(NsrDtsHandler
.SCALE_INSTANCE_XPATH
),
3486 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY| rwdts
.Flag
.CACHE
,
3489 self
._key
_pair
_regh
= acg
.register(
3490 xpath
=self
._project
.add_project(NsrDtsHandler
.KEY_PAIR_XPATH
),
3491 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3494 def deregister(self
):
3495 self
._log
.debug("De-register NSR config for project {}".
3496 format(self
._project
.name
))
3498 self
._nsr
_regh
.deregister()
3499 self
._nsr
_regh
= None
3500 if self
._scale
_regh
:
3501 self
._scale
_regh
.deregister()
3502 self
._scale
_regh
= None
3503 if self
._key
_pair
_regh
:
3504 self
._key
_pair
_regh
.deregister()
3505 self
._key
_pair
_regh
= None
3508 class NsrOpDataDtsHandler(object):
3509 """ The network service op data DTS handler """
3510 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
3512 def __init__(self
, dts
, log
, loop
, nsm
):
3518 self
._project
= nsm
._project
3523 """ Return the registration handle"""
3528 """ Return the NS manager instance """
3533 """ Register for Nsr op data publisher registration"""
3535 self
._log
.warning("NSR op data handler already registered for project {}".
3536 format(self
._project
.name
))
3539 xpath
= self
._project
.add_project(NsrOpDataDtsHandler
.XPATH
)
3540 self
._log
.debug("Registering Nsr op data path {} as publisher".
3543 hdl
= rift
.tasklets
.DTS
.RegistrationHandler()
3544 handlers
= rift
.tasklets
.Group
.Handler()
3545 with self
._dts
.group_create(handler
=handlers
) as group
:
3546 self
._regh
= group
.register(xpath
=xpath
,
3548 flags
=rwdts
.Flag
.PUBLISHER | rwdts
.Flag
.NO_PREP_READ | rwdts
.Flag
.DATASTORE
)
3550 def deregister(self
):
3551 self
._log
.debug("De-register NSR opdata for project {}".
3552 format(self
._project
.name
))
3554 self
._regh
.deregister()
3558 def create(self
, xpath
, msg
):
3560 Create an NS record in DTS with the path and message
3562 path
= self
._project
.add_project(xpath
)
3563 self
._log
.debug("Creating NSR %s:%s", path
, msg
)
3564 self
.regh
.create_element(path
, msg
)
3565 self
._log
.debug("Created NSR, %s:%s", path
, msg
)
3568 def update(self
, xpath
, msg
, flags
=rwdts
.XactFlag
.REPLACE
):
3570 Update an NS record in DTS with the path and message
3572 path
= self
._project
.add_project(xpath
)
3573 self
._log
.debug("Updating NSR, %s:%s regh = %s", path
, msg
, self
.regh
)
3574 self
.regh
.update_element(path
, msg
, flags
)
3575 self
._log
.debug("Updated NSR, %s:%s", path
, msg
)
3578 def delete(self
, xpath
):
3580 Update an NS record in DTS with the path and message
3582 path
= self
._project
.add_project(xpath
)
3583 self
._log
.debug("Deleting NSR path:%s", path
)
3584 self
.regh
.delete_element(path
)
3585 self
._log
.debug("Deleted NSR path:%s", path
)
3588 class VnfrDtsHandler(object):
3589 """ The virtual network service DTS handler """
3590 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
3592 def __init__(self
, dts
, log
, loop
, nsm
):
3602 """ Return registration handle """
3607 """ Return the NS manager instance """
3612 """ Register for vnfr create/update/delete/ advises from dts """
3614 self
._log
.warning("VNFR DTS handler already registered for project {}".
3615 format(self
._project
.name
))
3619 def on_commit(xact_info
):
3620 """ The transaction has been committed """
3621 self
._log
.debug("Got vnfr commit (xact_info: %s)", xact_info
)
3622 return rwdts
.MemberRspCode
.ACTION_OK
3625 def on_prepare(xact_info
, action
, ks_path
, msg
):
3626 """ prepare callback from dts """
3627 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3629 "Got vnfr on_prepare cb (xact_info: %s, action: %s): %s:%s",
3630 xact_info
, action
, ks_path
, msg
3633 schema
= VnfrYang
.YangData_RwProject_Project_VnfrCatalog_Vnfr
.schema()
3634 path_entry
= schema
.keyspec_to_entry(ks_path
)
3635 if path_entry
.key00
.id not in self
._nsm
._vnfrs
:
3636 # Check if this is a monitoring param xpath
3637 if 'vnfr:monitoring-param' not in xpath
:
3638 self
._log
.error("%s request for non existent record path %s",
3640 xact_info
.respond_xpath(rwdts
.XactRspCode
.NA
, xpath
)
3644 if action
== rwdts
.QueryAction
.CREATE
or action
== rwdts
.QueryAction
.UPDATE
:
3645 yield from self
._nsm
.update_vnfr(msg
)
3646 elif action
== rwdts
.QueryAction
.DELETE
:
3647 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3648 self
._nsm
.delete_vnfr(path_entry
.key00
.id)
3650 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
, xpath
)
3652 self
._log
.debug("Registering for VNFR using xpath: %s",
3653 VnfrDtsHandler
.XPATH
,)
3655 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_commit
=on_commit
,
3656 on_prepare
=on_prepare
,)
3657 with self
._dts
.group_create() as group
:
3658 self
._regh
= group
.register(xpath
=self
._nsm
._project
.add_project(
3659 VnfrDtsHandler
.XPATH
),
3661 flags
=(rwdts
.Flag
.SUBSCRIBER
),)
3663 def deregister(self
):
3664 self
._log
.debug("De-register VNFR for project {}".
3665 format(self
._project
.name
))
3667 self
._regh
.deregister()
3670 class NsdRefCountDtsHandler(object):
3671 """ The NSD Ref Count DTS handler """
3672 XPATH
= "D,/nsr:ns-instance-opdata/rw-nsr:nsd-ref-count"
3674 def __init__(self
, dts
, log
, loop
, nsm
):
3684 """ Return registration handle """
3689 """ Return the NS manager instance """
3694 """ Register for NSD ref count read from dts """
3696 self
._log
.warning("NSD ref DTS handler already registered for project {}".
3697 format(self
._project
.name
))
3702 def on_prepare(xact_info
, action
, ks_path
, msg
):
3703 """ prepare callback from dts """
3704 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3706 if action
== rwdts
.QueryAction
.READ
:
3707 schema
= RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount
.schema()
3708 path_entry
= schema
.keyspec_to_entry(ks_path
)
3709 nsd_list
= yield from self
._nsm
.get_nsd_refcount(path_entry
.key00
.nsd_id_ref
)
3710 for xpath
, msg
in nsd_list
:
3711 xact_info
.respond_xpath(rsp_code
=rwdts
.XactRspCode
.MORE
,
3714 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
3716 raise NetworkServiceRecordError("Not supported operation %s" % action
)
3718 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_prepare
,)
3719 with self
._dts
.group_create() as group
:
3720 self
._regh
= group
.register(xpath
=self
._nsm
._project
.add_project(
3721 NsdRefCountDtsHandler
.XPATH
),
3723 flags
=rwdts
.Flag
.PUBLISHER
,)
3725 def deregister(self
):
3726 self
._log
.debug("De-register NSD Ref count for project {}".
3727 format(self
._project
.name
))
3729 self
._regh
.deregister()
3733 class NsManager(object):
3734 """ The Network Service Manager class"""
3735 def __init__(self
, dts
, log
, loop
, project
,
3736 nsr_handler
, vnfr_handler
, vlr_handler
, ro_plugin_selector
,
3737 vnffgmgr
, vnfd_pub_handler
, cloud_account_handler
):
3741 self
._project
= project
3742 self
._nsr
_handler
= nsr_handler
3743 self
._vnfr
_pub
_handler
= vnfr_handler
3744 self
._vlr
_pub
_handler
= vlr_handler
3745 self
._vnffgmgr
= vnffgmgr
3746 self
._vnfd
_pub
_handler
= vnfd_pub_handler
3747 self
._cloud
_account
_handler
= cloud_account_handler
3749 self
._ro
_plugin
_selector
= ro_plugin_selector
3750 self
._ncclient
= rift
.mano
.ncclient
.NcClient(
3762 self
.cfgmgr_obj
= conman
.ROConfigManager(log
, loop
, dts
, self
)
3764 # TODO: All these handlers should move to tasklet level.
3765 # Passing self is often an indication of bad design
3766 self
._nsd
_dts
_handler
= NsdDtsHandler(dts
, log
, loop
, self
)
3767 self
._vnfd
_dts
_handler
= VnfdDtsHandler(dts
, log
, loop
, self
)
3768 self
._dts
_handlers
= [self
._nsd
_dts
_handler
,
3769 VnfrDtsHandler(dts
, log
, loop
, self
),
3770 NsdRefCountDtsHandler(dts
, log
, loop
, self
),
3771 NsrDtsHandler(dts
, log
, loop
, self
),
3772 ScalingRpcHandler(log
, dts
, loop
, self
._project
,
3773 self
.scale_rpc_callback
),
3774 NsrRpcDtsHandler(dts
, log
, loop
, self
),
3775 self
._vnfd
_dts
_handler
,
3796 def nsr_handler(self
):
3797 """" NSR handler """
3798 return self
._nsr
_handler
3802 """" So Obj handler """
3807 """ NSRs in this NSM"""
3812 """ NSDs in this NSM"""
3817 """ VNFDs in this NSM"""
3822 """ VNFRs in this NSM"""
3826 def nsr_pub_handler(self
):
3827 """ NSR publication handler """
3828 return self
._nsr
_handler
3831 def vnfr_pub_handler(self
):
3832 """ VNFR publication handler """
3833 return self
._vnfr
_pub
_handler
3836 def vlr_pub_handler(self
):
3837 """ VLR publication handler """
3838 return self
._vlr
_pub
_handler
3841 def vnfd_pub_handler(self
):
3842 return self
._vnfd
_pub
_handler
3846 """ Register all static DTS handlers """
3847 for dts_handle
in self
._dts
_handlers
:
3848 yield from dts_handle
.register()
3850 def deregister(self
):
3851 """ Register all static DTS handlers """
3852 for dts_handle
in self
._dts
_handlers
:
3853 yield from dts_handle
.deregister()
3856 def get_ns_by_nsr_id(self
, nsr_id
):
3857 """ get NSR by nsr id """
3858 if nsr_id
not in self
._nsrs
:
3859 raise NetworkServiceRecordError("NSR id %s not found" % nsr_id
)
3861 return self
._nsrs
[nsr_id
]
3863 def scale_nsr_out(self
, nsr_id
, scale_group_name
, instance_id
, config_xact
):
3864 self
.log
.debug("Scale out NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3869 nsr
= self
._nsrs
[nsr_id
]
3870 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3871 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3873 self
._loop
.create_task(nsr
.create_scale_group_instance(scale_group_name
, instance_id
, config_xact
))
3875 def scale_nsr_in(self
, nsr_id
, scale_group_name
, instance_id
):
3876 self
.log
.debug("Scale in NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3881 nsr
= self
._nsrs
[nsr_id
]
3882 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3883 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3885 self
._loop
.create_task(nsr
.delete_scale_group_instance(scale_group_name
, instance_id
))
3887 def scale_rpc_callback(self
, xact
, msg
, action
):
3888 """Callback handler for RPC calls
3890 xact : Transaction Handler
3892 action : Scaling Action
3894 ScalingGroupInstance
= NsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup_Instance
3895 ScalingGroup
= NsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup
3897 xpath
= self
._project
.add_project(
3898 ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').
3899 format(msg
.nsr_id_ref
))
3901 instance
= ScalingGroupInstance
.from_dict({
3902 "id": msg
.instance_id
,
3903 "project_name": self
._project
.name
,})
3906 def get_nsr_scaling_group():
3907 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3909 for result
in results
:
3910 res
= yield from result
3911 nsr_config
= res
.result
3913 for scaling_group
in nsr_config
.scaling_group
:
3914 if scaling_group
.scaling_group_name_ref
== msg
.scaling_group_name_ref
:
3917 scaling_group
= nsr_config
.scaling_group
.add()
3918 scaling_group
.scaling_group_name_ref
= msg
.scaling_group_name_ref
3920 return (nsr_config
, scaling_group
)
3923 def update_config(nsr_config
):
3924 xml
= self
._ncclient
.convert_to_xml(RwNsrYang
, nsr_config
)
3925 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
3926 yield from self
._ncclient
.connect()
3927 yield from self
._ncclient
.manager
.edit_config(target
="running", config
=xml
, default_operation
="replace")
3931 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3932 scaling_group
.instance
.append(instance
)
3933 yield from update_config(nsr_config
)
3937 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3938 scaling_group
.instance
.remove(instance
)
3939 yield from update_config(nsr_config
)
3941 if action
== ScalingRpcHandler
.ACTION
.SCALE_OUT
:
3942 self
._loop
.create_task(scale_out())
3944 self
._loop
.create_task(scale_in())
3946 # Opdata based calls, disabled for now!
3947 # if action == ScalingRpcHandler.ACTION.SCALE_OUT:
3948 # self.scale_nsr_out(
3950 # msg.scaling_group_name_ref,
3954 # self.scale_nsr_in(
3956 # msg.scaling_group_name_ref,
3959 def nsr_update_cfg(self
, nsr_id
, msg
):
3960 nsr
= self
._nsrs
[nsr_id
]
3961 nsr
.nsr_cfg_msg
= msg
3963 def nsr_instantiate_vl(self
, nsr_id
, vld
):
3964 self
.log
.debug("NSR {} create VL {}".format(nsr_id
, vld
))
3965 nsr
= self
._nsrs
[nsr_id
]
3966 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3967 raise NsrVlUpdateError("Cannot perform VL instantiate if NSR is not in running state")
3969 # Not calling in a separate task as this is called from a separate task
3970 yield from nsr
.create_vl_instance(vld
)
3972 def nsr_terminate_vl(self
, nsr_id
, vld
):
3973 self
.log
.debug("NSR {} delete VL {}".format(nsr_id
, vld
.id))
3974 nsr
= self
._nsrs
[nsr_id
]
3975 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3976 raise NsrVlUpdateError("Cannot perform VL terminate if NSR is not in running state")
3978 # Not calling in a separate task as this is called from a separate task
3979 yield from nsr
.delete_vl_instance(vld
)
3981 def create_nsr(self
, nsr_msg
, key_pairs
=None,restart_mode
=False):
3982 """ Create an NSR instance """
3983 if nsr_msg
.id in self
._nsrs
:
3984 msg
= "NSR id %s already exists" % nsr_msg
.id
3985 self
._log
.error(msg
)
3986 raise NetworkServiceRecordError(msg
)
3988 self
._log
.info("Create NetworkServiceRecord nsr id %s from nsd_id %s",
3992 nsm_plugin
= self
._ro
_plugin
_selector
.ro_plugin
3993 sdn_account_name
= self
._cloud
_account
_handler
.get_cloud_account_sdn_name(nsr_msg
.cloud_account
)
3995 nsr
= NetworkServiceRecord(self
._dts
,
4004 restart_mode
=restart_mode
,
4005 vlr_handler
=self
._ro
_plugin
_selector
._records
_publisher
._vlr
_pub
_hdlr
4007 self
._nsrs
[nsr_msg
.id] = nsr
4008 nsm_plugin
.create_nsr(nsr_msg
, nsr_msg
.nsd
, key_pairs
)
4012 def delete_nsr(self
, nsr_id
):
4014 Delete NSR with the passed nsr id
4016 del self
._nsrs
[nsr_id
]
4019 def instantiate_ns(self
, nsr_id
, config_xact
):
4020 """ Instantiate an NS instance """
4021 self
._log
.debug("Instantiating Network service id %s", nsr_id
)
4022 if nsr_id
not in self
._nsrs
:
4023 err
= "NSR id %s not found " % nsr_id
4024 self
._log
.error(err
)
4025 raise NetworkServiceRecordError(err
)
4027 nsr
= self
._nsrs
[nsr_id
]
4028 yield from nsr
.nsm_plugin
.instantiate_ns(nsr
, config_xact
)
4031 def update_vnfr(self
, vnfr
):
4032 """Create/Update an VNFR """
4034 vnfr_state
= self
._vnfrs
[vnfr
.id].state
4035 self
._log
.debug("Updating VNFR with state %s: vnfr %s", vnfr_state
, vnfr
)
4037 yield from self
._vnfrs
[vnfr
.id].update_state(vnfr
)
4038 nsr
= self
.find_nsr_for_vnfr(vnfr
.id)
4039 yield from nsr
.update_state()
4041 def find_nsr_for_vnfr(self
, vnfr_id
):
4042 """ Find the NSR which )has the passed vnfr id"""
4043 for nsr
in list(self
.nsrs
.values()):
4044 for vnfr
in list(nsr
.vnfrs
.values()):
4045 if vnfr
.id == vnfr_id
:
4049 def delete_vnfr(self
, vnfr_id
):
4050 """ Delete VNFR with the passed id"""
4051 del self
._vnfrs
[vnfr_id
]
4053 def get_nsd_ref(self
, nsd_id
):
4054 """ Get network service descriptor for the passed nsd_id
4056 nsd
= self
.get_nsd(nsd_id
)
4061 def get_nsr_config(self
, nsd_id
):
4062 xpath
= self
._project
.add_project("C,/nsr:ns-instance-config")
4063 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
4065 for result
in results
:
4066 entry
= yield from result
4067 ns_instance_config
= entry
.result
4069 for nsr
in ns_instance_config
.nsr
:
4070 if nsr
.nsd
.id == nsd_id
:
4076 def nsd_unref_by_nsr_id(self
, nsr_id
):
4077 """ Unref the network service descriptor based on NSR id """
4078 self
._log
.debug("NSR Unref called for Nsr Id:%s", nsr_id
)
4079 if nsr_id
in self
._nsrs
:
4080 nsr
= self
._nsrs
[nsr_id
]
4083 nsd
= self
.get_nsd(nsr
.nsd_id
)
4084 self
._log
.debug("Releasing ref on NSD %s held by NSR %s - Curr %d",
4085 nsd
.id, nsr
.id, nsd
.ref_count
)
4087 except NetworkServiceDescriptorError
:
4088 # We store a copy of NSD in NSR and the NSD in nsd-catalog
4093 self
._log
.error("Cannot find NSR with id %s", nsr_id
)
4094 raise NetworkServiceDescriptorUnrefError("No NSR with id" % nsr_id
)
4097 def nsd_unref(self
, nsd_id
):
4098 """ Unref the network service descriptor associated with the id """
4099 nsd
= self
.get_nsd(nsd_id
)
4102 def get_nsd(self
, nsd_id
):
4103 """ Get network service descriptor for the passed nsd_id"""
4104 if nsd_id
not in self
._nsds
:
4105 self
._log
.error("Cannot find NSD id:%s", nsd_id
)
4106 raise NetworkServiceDescriptorError("Cannot find NSD id:%s", nsd_id
)
4108 return self
._nsds
[nsd_id
]
4110 def create_nsd(self
, nsd_msg
):
4111 """ Create a network service descriptor """
4112 self
._log
.debug("Create network service descriptor - %s", nsd_msg
)
4113 if nsd_msg
.id in self
._nsds
:
4114 self
._log
.error("Cannot create NSD %s -NSD ID already exists", nsd_msg
)
4115 raise NetworkServiceDescriptorError("NSD already exists-%s", nsd_msg
.id)
4117 nsd
= NetworkServiceDescriptor(
4124 self
._nsds
[nsd_msg
.id] = nsd
4128 def update_nsd(self
, nsd
):
4129 """ update the Network service descriptor """
4130 self
._log
.debug("Update network service descriptor - %s", nsd
)
4131 if nsd
.id not in self
._nsds
:
4132 self
._log
.debug("No NSD found - creating NSD id = %s", nsd
.id)
4133 self
.create_nsd(nsd
)
4135 self
._log
.debug("Updating NSD id = %s, nsd = %s", nsd
.id, nsd
)
4136 self
._nsds
[nsd
.id].update(nsd
)
4138 def delete_nsd(self
, nsd_id
):
4139 """ Delete the Network service descriptor with the passed id """
4140 self
._log
.debug("Deleting the network service descriptor - %s", nsd_id
)
4141 if nsd_id
not in self
._nsds
:
4142 self
._log
.debug("Delete NSD failed - cannot find nsd-id %s", nsd_id
)
4143 raise NetworkServiceDescriptorNotFound("Cannot find %s", nsd_id
)
4145 if nsd_id
not in self
._nsds
:
4146 self
._log
.debug("Cannot delete NSD id %s reference exists %s",
4148 self
._nsds
[nsd_id
].ref_count
)
4149 raise NetworkServiceDescriptorRefCountExists(
4150 "Cannot delete :%s, ref_count:%s",
4152 self
._nsds
[nsd_id
].ref_count
)
4154 del self
._nsds
[nsd_id
]
4156 def get_vnfd_config(self
, xact
):
4157 vnfd_dts_reg
= self
._vnfd
_dts
_handler
.regh
4158 for cfg
in vnfd_dts_reg
.get_xact_elements(xact
):
4159 if cfg
.id not in self
._vnfds
:
4160 self
.create_vnfd(cfg
)
4162 def get_vnfd(self
, vnfd_id
, xact
):
4163 """ Get virtual network function descriptor for the passed vnfd_id"""
4164 if vnfd_id
not in self
._vnfds
:
4165 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4166 self
.get_vnfd_config(xact
)
4168 if vnfd_id
not in self
._vnfds
:
4169 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4170 raise VnfDescriptorError("Cannot find VNFD id:%s", vnfd_id
)
4172 return self
._vnfds
[vnfd_id
]
4174 def create_vnfd(self
, vnfd
):
4175 """ Create a virtual network function descriptor """
4176 self
._log
.debug("Create virtual network function descriptor - %s", vnfd
)
4177 if vnfd
.id in self
._vnfds
:
4178 self
._log
.error("Cannot create VNFD %s -VNFD ID already exists", vnfd
)
4179 raise VnfDescriptorError("VNFD already exists-%s", vnfd
.id)
4181 self
._vnfds
[vnfd
.id] = vnfd
4182 return self
._vnfds
[vnfd
.id]
4184 def update_vnfd(self
, vnfd
):
4185 """ Update the virtual network function descriptor """
4186 self
._log
.debug("Update virtual network function descriptor- %s", vnfd
)
4189 if vnfd
.id not in self
._vnfds
:
4190 self
._log
.debug("No VNFD found - creating VNFD id = %s", vnfd
.id)
4191 self
.create_vnfd(vnfd
)
4193 self
._log
.debug("Updating VNFD id = %s, vnfd = %s", vnfd
.id, vnfd
)
4194 self
._vnfds
[vnfd
.id] = vnfd
4197 def delete_vnfd(self
, vnfd_id
):
4198 """ Delete the virtual network function descriptor with the passed id """
4199 self
._log
.debug("Deleting the virtual network function descriptor - %s", vnfd_id
)
4200 if vnfd_id
not in self
._vnfds
:
4201 self
._log
.debug("Delete VNFD failed - cannot find vnfd-id %s", vnfd_id
)
4202 raise VnfDescriptorError("Cannot find %s", vnfd_id
)
4204 del self
._vnfds
[vnfd_id
]
4206 def nsd_in_use(self
, nsd_id
):
4207 """ Is the NSD with the passed id in use """
4208 self
._log
.debug("Is this NSD in use - msg:%s", nsd_id
)
4209 if nsd_id
in self
._nsds
:
4210 return self
._nsds
[nsd_id
].in_use()
4214 def publish_nsr(self
, xact
, path
, msg
):
4215 """ Publish a NSR """
4216 self
._log
.debug("Publish NSR with path %s, msg %s",
4218 yield from self
.nsr_handler
.update(xact
, path
, msg
)
4221 def unpublish_nsr(self
, xact
, path
):
4222 """ Un Publish an NSR """
4223 self
._log
.debug("Publishing delete NSR with path %s", path
)
4224 yield from self
.nsr_handler
.delete(path
, xact
)
4226 def vnfr_is_ready(self
, vnfr_id
):
4227 """ VNFR with the id is ready """
4228 self
._log
.debug("VNFR id %s ready", vnfr_id
)
4229 if vnfr_id
not in self
._vnfds
:
4230 err
= "Did not find VNFR ID with id %s" % vnfr_id
4231 self
._log
.critical("err")
4232 raise VirtualNetworkFunctionRecordError(err
)
4233 self
._vnfrs
[vnfr_id
].is_ready()
4236 def get_nsd_refcount(self
, nsd_id
):
4237 """ Get the nsd_list from this NSM"""
4239 def nsd_refcount_xpath(nsd_id
):
4240 """ xpath for ref count entry """
4241 return (self
._project
.add_project(NsdRefCountDtsHandler
.XPATH
) +
4242 "[rw-nsr:nsd-id-ref = '{}']").format(nsd_id
)
4245 if nsd_id
is None or nsd_id
== "":
4246 for nsd
in self
._nsds
.values():
4247 nsd_msg
= RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount()
4248 nsd_msg
.nsd_id_ref
= nsd
.id
4249 nsd_msg
.instance_ref_count
= nsd
.ref_count
4250 nsd_list
.append((nsd_refcount_xpath(nsd
.id), nsd_msg
))
4251 elif nsd_id
in self
._nsds
:
4252 nsd_msg
= RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount()
4253 nsd_msg
.nsd_id_ref
= self
._nsds
[nsd_id
].id
4254 nsd_msg
.instance_ref_count
= self
._nsds
[nsd_id
].ref_count
4255 nsd_list
.append((nsd_refcount_xpath(nsd_id
), nsd_msg
))
4260 def terminate_ns(self
, nsr_id
, xact
):
4262 Terminate network service for the given NSR Id
4265 # Terminate the instances/networks assocaited with this nw service
4266 self
._log
.debug("Terminating the network service %s", nsr_id
)
4268 yield from self
._nsrs
[nsr_id
].terminate()
4269 except Exception as e
:
4270 self
.log
.exception("Failed to terminate NSR[id=%s]", nsr_id
)
4273 yield from self
.nsd_unref_by_nsr_id(nsr_id
)
4275 # Unpublish the NSR record
4276 self
._log
.debug("Unpublishing the network service %s", nsr_id
)
4277 yield from self
._nsrs
[nsr_id
].unpublish(xact
)
4279 # Finaly delete the NS instance from this NS Manager
4280 self
._log
.debug("Deletng the network service %s", nsr_id
)
4281 self
.delete_nsr(nsr_id
)
4284 class NsmRecordsPublisherProxy(object):
4285 """ This class provides a publisher interface that allows plugin objects
4286 to publish NSR/VNFR/VLR"""
4288 def __init__(self
, dts
, log
, loop
, project
, nsr_pub_hdlr
,
4289 vnfr_pub_hdlr
, vlr_pub_hdlr
,):
4293 self
._project
= project
4294 self
._nsr
_pub
_hdlr
= nsr_pub_hdlr
4295 self
._vlr
_pub
_hdlr
= vlr_pub_hdlr
4296 self
._vnfr
_pub
_hdlr
= vnfr_pub_hdlr
4299 def publish_nsr(self
, xact
, nsr
):
4300 """ Publish an NSR """
4301 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4302 return (yield from self
._nsr
_pub
_hdlr
.update(xact
, path
, nsr
))
4305 def unpublish_nsr(self
, xact
, nsr
):
4306 """ Unpublish an NSR """
4307 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4308 return (yield from self
._nsr
_pub
_hdlr
.delete(xact
, path
))
4311 def publish_vnfr(self
, xact
, vnfr
):
4312 """ Publish an VNFR """
4313 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4314 return (yield from self
._vnfr
_pub
_hdlr
.update(xact
, path
, vnfr
))
4317 def unpublish_vnfr(self
, xact
, vnfr
):
4318 """ Unpublish a VNFR """
4319 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4320 return (yield from self
._vnfr
_pub
_hdlr
.delete(xact
, path
))
4323 def publish_vlr(self
, xact
, vlr
):
4324 """ Publish a VLR """
4325 path
= self
._project
.add_project(VirtualLinkRecord
.vlr_xpath(vlr
))
4326 return (yield from self
._vlr
_pub
_hdlr
.update(xact
, path
, vlr
))
4329 def unpublish_vlr(self
, xact
, vlr
):
4330 """ Unpublish a VLR """
4331 path
= self
._project
.add_project(VirtualLinkRecord
.vlr_xpath(vlr
))
4332 return (yield from self
._vlr
_pub
_hdlr
.delete(xact
, path
))
4335 class ScalingRpcHandler(mano_dts
.DtsHandler
):
4336 """ The Network service Monitor DTS handler """
4337 SCALE_IN_INPUT_XPATH
= "I,/nsr:exec-scale-in"
4338 SCALE_IN_OUTPUT_XPATH
= "O,/nsr:exec-scale-in"
4340 SCALE_OUT_INPUT_XPATH
= "I,/nsr:exec-scale-out"
4341 SCALE_OUT_OUTPUT_XPATH
= "O,/nsr:exec-scale-out"
4343 ACTION
= Enum('ACTION', 'SCALE_IN SCALE_OUT')
4345 def __init__(self
, log
, dts
, loop
, project
, callback
=None):
4346 super().__init
__(log
, dts
, loop
, project
)
4347 self
.callback
= callback
4348 self
.last_instance_id
= defaultdict(int)
4349 self
._regh
_in
= None
4350 self
._regh
_out
= None
4356 self
._log
.warning("RPC already registered for project {}".
4357 format(self
._project
.name
))
4361 def on_scale_in_prepare(xact_info
, action
, ks_path
, msg
):
4362 assert action
== rwdts
.QueryAction
.RPC
4365 if not self
._project
.rpc_check(msg
, xact_info
=xact_info
):
4369 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_IN
)
4371 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleIn
.from_dict({
4372 "instance_id": msg
.instance_id
,
4373 "project_name": self
._project
.name
,})
4375 xact_info
.respond_xpath(
4376 rwdts
.XactRspCode
.ACK
,
4377 self
.__class
__.SCALE_IN_OUTPUT_XPATH
,
4380 except Exception as e
:
4381 self
.log
.exception(e
)
4382 xact_info
.respond_xpath(
4383 rwdts
.XactRspCode
.NACK
,
4384 self
.__class
__.SCALE_IN_OUTPUT_XPATH
)
4387 def on_scale_out_prepare(xact_info
, action
, ks_path
, msg
):
4388 assert action
== rwdts
.QueryAction
.RPC
4391 if not self
._project
.rpc_check(msg
, xact_info
=xact_info
):
4394 scaling_group
= msg
.scaling_group_name_ref
4395 if not msg
.instance_id
:
4396 last_instance_id
= self
.last_instance_id
[scale_group
]
4397 msg
.instance_id
= last_instance_id
+ 1
4398 self
.last_instance_id
[scale_group
] += 1
4401 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_OUT
)
4403 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleOut
.from_dict({
4404 "instance_id": msg
.instance_id
,
4405 "project_name": self
._project
.name
,})
4407 xact_info
.respond_xpath(
4408 rwdts
.XactRspCode
.ACK
,
4409 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
,
4412 except Exception as e
:
4413 self
.log
.exception(e
)
4414 xact_info
.respond_xpath(
4415 rwdts
.XactRspCode
.NACK
,
4416 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
)
4418 scale_in_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4419 on_prepare
=on_scale_in_prepare
)
4420 scale_out_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4421 on_prepare
=on_scale_out_prepare
)
4423 with self
.dts
.group_create() as group
:
4424 self
._regh
_in
= group
.register(
4425 xpath
=self
.__class
__.SCALE_IN_INPUT_XPATH
,
4426 handler
=scale_in_hdl
,
4427 flags
=rwdts
.Flag
.PUBLISHER
)
4428 self
._regh
_out
= group
.register(
4429 xpath
=self
.__class
__.SCALE_OUT_INPUT_XPATH
,
4430 handler
=scale_out_hdl
,
4431 flags
=rwdts
.Flag
.PUBLISHER
)
4433 def deregister(self
):
4434 self
._log
.debug("De-register scale RPCs for project {}".
4435 format(self
._project
.name
))
4437 self
._regh
_in
.deregister()
4438 self
._regh
_in
= None
4440 self
._regh
_out
.deregister()
4441 self
._regh
_out
= None
4444 class NsmProject(ManoProject
):
4446 def __init__(self
, name
, tasklet
, **kw
):
4447 super(NsmProject
, self
).__init
__(tasklet
.log
, name
)
4448 self
.update(tasklet
)
4452 self
._ro
_plugin
_selector
= None
4453 self
._vnffgmgr
= None
4455 self
._nsr
_pub
_handler
= None
4456 self
._vnfr
_pub
_handler
= None
4457 self
._vlr
_pub
_handler
= None
4458 self
._vnfd
_pub
_handler
= None
4459 self
._scale
_cfg
_handler
= None
4461 self
._records
_publisher
_proxy
= None
4465 self
._nsr
_pub
_handler
= publisher
.NsrOpDataDtsHandler(
4466 self
._dts
, self
.log
, self
.loop
, self
)
4467 yield from self
._nsr
_pub
_handler
.register()
4469 self
._vnfr
_pub
_handler
= publisher
.VnfrPublisherDtsHandler(
4470 self
._dts
, self
.log
, self
.loop
, self
)
4471 yield from self
._vnfr
_pub
_handler
.register()
4473 self
._vlr
_pub
_handler
= publisher
.VlrPublisherDtsHandler(
4474 self
._dts
, self
.log
, self
.loop
, self
)
4475 yield from self
._vlr
_pub
_handler
.register()
4477 manifest
= self
._tasklet
.tasklet_info
.get_pb_manifest()
4478 use_ssl
= manifest
.bootstrap_phase
.rwsecurity
.use_ssl
4479 ssl_cert
= manifest
.bootstrap_phase
.rwsecurity
.cert
4480 ssl_key
= manifest
.bootstrap_phase
.rwsecurity
.key
4482 self
._vnfd
_pub
_handler
= publisher
.VnfdPublisher(
4483 use_ssl
, ssl_cert
, ssl_key
, self
.loop
, self
)
4485 self
._records
_publisher
_proxy
= NsmRecordsPublisherProxy(
4490 self
._nsr
_pub
_handler
,
4491 self
._vnfr
_pub
_handler
,
4492 self
._vlr
_pub
_handler
,
4495 # Register the NSM to receive the nsm plugin
4496 # when cloud account is configured
4497 self
._ro
_plugin
_selector
= cloud
.ROAccountPluginSelector(
4502 self
._records
_publisher
_proxy
,
4504 yield from self
._ro
_plugin
_selector
.register()
4506 self
._cloud
_account
_handler
= cloud
.CloudAccountConfigSubscriber(
4513 yield from self
._cloud
_account
_handler
.register()
4515 self
._vnffgmgr
= rwvnffgmgr
.VnffgMgr(self
._dts
, self
.log
, self
.log_hdl
, self
.loop
, self
)
4516 yield from self
._vnffgmgr
.register()
4518 self
._nsm
= NsManager(
4523 self
._nsr
_pub
_handler
,
4524 self
._vnfr
_pub
_handler
,
4525 self
._vlr
_pub
_handler
,
4526 self
._ro
_plugin
_selector
,
4528 self
._vnfd
_pub
_handler
,
4529 self
._cloud
_account
_handler
,
4532 yield from self
._nsm
.register()
4534 def deregister(self
):
4535 self
._log
.debug("Project {} de-register".format(self
.name
))
4536 self
._nsm
.deregister()
4537 self
._vnffgmgr
.deregister()
4538 self
._cloud
_account
_handler
.deregister()
4539 self
._ro
_plugin
_selector
.deregister()
4542 class NsmTasklet(rift
.tasklets
.Tasklet
):
4544 The network service manager tasklet
4546 def __init__(self
, *args
, **kwargs
):
4547 super(NsmTasklet
, self
).__init
__(*args
, **kwargs
)
4548 self
.rwlog
.set_category("rw-mano-log")
4549 self
.rwlog
.set_subcategory("nsm")
4552 self
.project_handler
= None
4560 """ The task start callback """
4561 super(NsmTasklet
, self
).start()
4562 self
.log
.info("Starting NsmTasklet")
4564 self
.log
.debug("Registering with dts")
4565 self
._dts
= rift
.tasklets
.DTS(self
.tasklet_info
,
4566 RwNsmYang
.get_schema(),
4568 self
.on_dts_state_change
)
4570 self
.log
.debug("Created DTS Api GI Object: %s", self
._dts
)
4576 print("Caught Exception in NSM stop:", sys
.exc_info()[0])
4579 def on_instance_started(self
):
4580 """ Task instance started callback """
4581 self
.log
.debug("Got instance started callback")
4585 """ Task init callback """
4586 self
.log
.debug("Got instance started callback")
4588 self
.log
.debug("creating project handler")
4589 self
.project_handler
= ProjectHandler(self
, NsmProject
)
4590 self
.project_handler
.register()
4596 """ Task run callback """
4600 def on_dts_state_change(self
, state
):
4601 """Take action according to current dts state to transition
4602 application into the corresponding application state
4605 state - current dts state
4608 rwdts
.State
.INIT
: rwdts
.State
.REGN_COMPLETE
,
4609 rwdts
.State
.CONFIG
: rwdts
.State
.RUN
,
4613 rwdts
.State
.INIT
: self
.init
,
4614 rwdts
.State
.RUN
: self
.run
,
4617 # Transition application to next state
4618 handler
= handlers
.get(state
, None)
4619 if handler
is not None:
4620 yield from handler()
4622 # Transition dts to next state
4623 next_state
= switch
.get(state
, None)
4624 if next_state
is not None:
4625 self
.log
.debug("Changing state to %s", next_state
)
4626 self
._dts
.handle
.set_state(next_state
)