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('RwNsdYang', '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 (
61 import rift
.mano
.ncclient
62 import rift
.mano
.config_data
.config
63 import rift
.mano
.dts
as mano_dts
65 from . import rwnsm_conman
as conman
67 from . import publisher
69 from . import config_value_pool
70 from . import rwvnffgmgr
71 from . import scale_group
74 class NetworkServiceRecordState(Enum
):
75 """ Network Service Record State """
79 VNFFG_INIT_PHASE
= 104
85 VL_TERMINATE_PHASE
= 111
86 VNF_TERMINATE_PHASE
= 112
87 VNFFG_TERMINATE_PHASE
= 113
94 class NetworkServiceRecordError(Exception):
95 """ Network Service Record Error """
99 class NetworkServiceDescriptorError(Exception):
100 """ Network Service Descriptor Error """
104 class VirtualNetworkFunctionRecordError(Exception):
105 """ Virtual Network Function Record Error """
109 class NetworkServiceDescriptorNotFound(Exception):
110 """ Cannot find Network Service Descriptor"""
114 class NetworkServiceDescriptorRefCountExists(Exception):
115 """ Network Service Descriptor reference count exists """
119 class NetworkServiceDescriptorUnrefError(Exception):
120 """ Failed to unref a network service descriptor """
124 class NsrInstantiationFailed(Exception):
125 """ Failed to instantiate network service """
129 class VnfInstantiationFailed(Exception):
130 """ Failed to instantiate virtual network function"""
134 class VnffgInstantiationFailed(Exception):
135 """ Failed to instantiate virtual network function"""
139 class VnfDescriptorError(Exception):
140 """Failed to instantiate virtual network function"""
144 class ScalingOperationError(Exception):
148 class ScaleGroupMissingError(Exception):
152 class PlacementGroupError(Exception):
156 class NsrNsdUpdateError(Exception):
160 class NsrVlUpdateError(NsrNsdUpdateError
):
164 class VlRecordState(Enum
):
165 """ VL Record State """
167 INSTANTIATION_PENDING
= 102
169 TERMINATE_PENDING
= 104
174 class VnffgRecordState(Enum
):
175 """ VNFFG Record State """
177 INSTANTIATION_PENDING
= 102
179 TERMINATE_PENDING
= 104
184 class VnffgRecord(object):
185 """ Vnffg Records class"""
188 def __init__(self
, dts
, log
, loop
, vnffgmgr
, nsr
, nsr_name
, vnffgd_msg
, sdn_account_name
):
193 self
._vnffgmgr
= vnffgmgr
195 self
._nsr
_name
= nsr_name
196 self
._vnffgd
_msg
= vnffgd_msg
197 if sdn_account_name
is None:
198 self
._sdn
_account
_name
= ''
200 self
._sdn
_account
_name
= sdn_account_name
202 self
._vnffgr
_id
= str(uuid
.uuid4())
203 self
._vnffgr
_rsp
_id
= list()
204 self
._vnffgr
_state
= VnffgRecordState
.INIT
209 return self
._vnffgr
_id
213 """ state of this VNF """
214 return self
._vnffgr
_state
216 def fetch_vnffgr(self
):
218 Get VNFFGR message to be published
221 if self
._vnffgr
_state
== VnffgRecordState
.INIT
:
222 vnffgr_dict
= {"id": self
._vnffgr
_id
,
223 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
224 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
225 "sdn_account": self
._sdn
_account
_name
,
226 "operational_status": 'init',
228 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
229 elif self
._vnffgr
_state
== VnffgRecordState
.TERMINATED
:
230 vnffgr_dict
= {"id": self
._vnffgr
_id
,
231 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
232 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
233 "sdn_account": self
._sdn
_account
_name
,
234 "operational_status": 'terminated',
236 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
239 vnffgr
= self
._vnffgmgr
.fetch_vnffgr(self
._vnffgr
_id
)
241 self
._log
.exception("Fetching VNFFGR for VNFFG with id %s failed", self
._vnffgr
_id
)
242 self
._vnffgr
_state
= VnffgRecordState
.FAILED
243 vnffgr_dict
= {"id": self
._vnffgr
_id
,
244 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
245 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
246 "sdn_account": self
._sdn
_account
_name
,
247 "operational_status": 'failed',
249 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
254 def vnffgr_create_msg(self
):
255 """ Virtual Link Record message for Creating VLR in VNS """
256 vnffgr_dict
= {"id": self
._vnffgr
_id
,
257 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
258 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
259 "sdn_account": self
._sdn
_account
_name
,
261 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
262 for rsp
in self
._vnffgd
_msg
.rsp
:
263 vnffgr_rsp
= vnffgr
.rsp
.add()
264 vnffgr_rsp
.id = str(uuid
.uuid4())
265 vnffgr_rsp
.name
= self
._nsr
.name
+ '.' + rsp
.name
266 self
._vnffgr
_rsp
_id
.append(vnffgr_rsp
.id)
267 vnffgr_rsp
.vnffgd_rsp_id_ref
= rsp
.id
268 vnffgr_rsp
.vnffgd_rsp_name_ref
= rsp
.name
269 for rsp_cp_ref
in rsp
.vnfd_connection_point_ref
:
270 vnfd
= [vnfr
.vnfd
for vnfr
in self
._nsr
.vnfrs
.values() if vnfr
.vnfd
.id == rsp_cp_ref
.vnfd_id_ref
]
271 self
._log
.debug("VNFD message during VNFFG instantiation is %s",vnfd
)
272 if len(vnfd
) > 0 and vnfd
[0].has_field('service_function_type'):
273 self
._log
.debug("Service Function Type for VNFD ID %s is %s",rsp_cp_ref
.vnfd_id_ref
, vnfd
[0].service_function_type
)
275 self
._log
.error("Service Function Type not available for VNFD ID %s; Skipping in chain",rsp_cp_ref
.vnfd_id_ref
)
278 vnfr_cp_ref
= vnffgr_rsp
.vnfr_connection_point_ref
.add()
279 vnfr_cp_ref
.member_vnf_index_ref
= rsp_cp_ref
.member_vnf_index_ref
280 vnfr_cp_ref
.hop_number
= rsp_cp_ref
.order
281 vnfr_cp_ref
.vnfd_id_ref
=rsp_cp_ref
.vnfd_id_ref
282 vnfr_cp_ref
.service_function_type
= vnfd
[0].service_function_type
283 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
284 if (nsr_vnfr
.vnfd
.id == vnfr_cp_ref
.vnfd_id_ref
and
285 nsr_vnfr
.member_vnf_index
== vnfr_cp_ref
.member_vnf_index_ref
):
286 vnfr_cp_ref
.vnfr_id_ref
= nsr_vnfr
.id
287 vnfr_cp_ref
.vnfr_name_ref
= nsr_vnfr
.name
288 vnfr_cp_ref
.vnfr_connection_point_ref
= rsp_cp_ref
.vnfd_connection_point_ref
290 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
291 self
._log
.debug(" Received VNFR is %s", vnfr
)
292 while vnfr
.operational_status
!= 'running':
293 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
294 if vnfr
.operational_status
== 'failed':
295 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
296 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self
.id, vnfr
.id))
297 yield from asyncio
.sleep(2, loop
=self
._loop
)
298 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
299 self
._log
.debug("Received VNFR is %s", vnfr
)
301 vnfr_cp_ref
.connection_point_params
.mgmt_address
= vnfr
.mgmt_interface
.ip_address
302 for cp
in vnfr
.connection_point
:
303 if cp
.name
== vnfr_cp_ref
.vnfr_connection_point_ref
:
304 vnfr_cp_ref
.connection_point_params
.port_id
= cp
.connection_point_id
305 vnfr_cp_ref
.connection_point_params
.name
= self
._nsr
.name
+ '.' + cp
.name
306 for vdu
in vnfr
.vdur
:
307 for ext_intf
in vdu
.external_interface
:
308 if ext_intf
.name
== vnfr_cp_ref
.vnfr_connection_point_ref
:
309 vnfr_cp_ref
.connection_point_params
.vm_id
= vdu
.vim_id
310 self
._log
.debug("VIM ID for CP %s in VNFR %s is %s",cp
.name
,nsr_vnfr
.id,
311 vnfr_cp_ref
.connection_point_params
.vm_id
)
314 vnfr_cp_ref
.connection_point_params
.address
= cp
.ip_address
315 vnfr_cp_ref
.connection_point_params
.port
= VnffgRecord
.SFF_DP_PORT
317 for vnffgd_classifier
in self
._vnffgd
_msg
.classifier
:
318 _rsp
= [rsp
for rsp
in vnffgr
.rsp
if rsp
.vnffgd_rsp_id_ref
== vnffgd_classifier
.rsp_id_ref
]
320 rsp_id_ref
= _rsp
[0].id
321 rsp_name
= _rsp
[0].name
323 self
._log
.error("RSP with ID %s not found during classifier creation for classifier id %s",vnffgd_classifier
.rsp_id_ref
,vnffgd_classifier
.id)
325 vnffgr_classifier
= vnffgr
.classifier
.add()
326 vnffgr_classifier
.id = vnffgd_classifier
.id
327 vnffgr_classifier
.name
= self
._nsr
.name
+ '.' + vnffgd_classifier
.name
328 _rsp
[0].classifier_name
= vnffgr_classifier
.name
329 vnffgr_classifier
.rsp_id_ref
= rsp_id_ref
330 vnffgr_classifier
.rsp_name
= rsp_name
331 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
332 if (nsr_vnfr
.vnfd
.id == vnffgd_classifier
.vnfd_id_ref
and
333 nsr_vnfr
.member_vnf_index
== vnffgd_classifier
.member_vnf_index_ref
):
334 vnffgr_classifier
.vnfr_id_ref
= nsr_vnfr
.id
335 vnffgr_classifier
.vnfr_name_ref
= nsr_vnfr
.name
336 vnffgr_classifier
.vnfr_connection_point_ref
= vnffgd_classifier
.vnfd_connection_point_ref
338 if nsr_vnfr
.vnfd
.service_function_chain
== 'CLASSIFIER':
339 vnffgr_classifier
.sff_name
= nsr_vnfr
.name
341 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
342 self
._log
.debug(" Received VNFR is %s", vnfr
)
343 while vnfr
.operational_status
!= 'running':
344 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
345 if vnfr
.operational_status
== 'failed':
346 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
347 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self
.id, vnfr
.id))
348 yield from asyncio
.sleep(2, loop
=self
._loop
)
349 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
350 self
._log
.debug("Received VNFR is %s", vnfr
)
352 for cp
in vnfr
.connection_point
:
353 if cp
.name
== vnffgr_classifier
.vnfr_connection_point_ref
:
354 vnffgr_classifier
.port_id
= cp
.connection_point_id
355 vnffgr_classifier
.ip_address
= cp
.ip_address
356 for vdu
in vnfr
.vdur
:
357 for ext_intf
in vdu
.external_interface
:
358 if ext_intf
.name
== vnffgr_classifier
.vnfr_connection_point_ref
:
359 vnffgr_classifier
.vm_id
= vdu
.vim_id
360 self
._log
.debug("VIM ID for CP %s in VNFR %s is %s",cp
.name
,nsr_vnfr
.id,
361 vnfr_cp_ref
.connection_point_params
.vm_id
)
364 self
._log
.info("VNFFGR msg to be sent is %s", vnffgr
)
368 def vnffgr_nsr_sff_list(self
):
369 """ SFF List for VNFR """
371 sf_list
= [nsr_vnfr
.name
for nsr_vnfr
in self
._nsr
.vnfrs
.values() if nsr_vnfr
.vnfd
.service_function_chain
== 'SF']
373 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
374 if (nsr_vnfr
.vnfd
.service_function_chain
== 'CLASSIFIER' or nsr_vnfr
.vnfd
.service_function_chain
== 'SFF'):
375 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
376 self
._log
.debug(" Received VNFR is %s", vnfr
)
377 while vnfr
.operational_status
!= 'running':
378 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
379 if vnfr
.operational_status
== 'failed':
380 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
381 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self
.id, vnfr
.id))
382 yield from asyncio
.sleep(2, loop
=self
._loop
)
383 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
384 self
._log
.debug("Received VNFR is %s", vnfr
)
386 sff
= RwsdnYang
.VNFFGSff()
387 sff_list
[nsr_vnfr
.vnfd
.id] = sff
388 sff
.name
= nsr_vnfr
.name
389 sff
.function_type
= nsr_vnfr
.vnfd
.service_function_chain
391 sff
.mgmt_address
= vnfr
.mgmt_interface
.ip_address
392 sff
.mgmt_port
= VnffgRecord
.SFF_MGMT_PORT
393 for cp
in vnfr
.connection_point
:
394 sff_dp
= sff
.dp_endpoints
.add()
395 sff_dp
.name
= self
._nsr
.name
+ '.' + cp
.name
396 sff_dp
.address
= cp
.ip_address
397 sff_dp
.port
= VnffgRecord
.SFF_DP_PORT
398 if nsr_vnfr
.vnfd
.service_function_chain
== 'SFF':
399 for sf_name
in sf_list
:
400 _sf
= sff
.vnfr_list
.add()
401 _sf
.vnfr_name
= sf_name
406 def instantiate(self
):
407 """ Instantiate this VNFFG """
409 self
._log
.info("Instaniating VNFFGR with vnffgd %s",
413 vnffgr_request
= yield from self
.vnffgr_create_msg()
414 vnffg_sff_list
= yield from self
.vnffgr_nsr_sff_list()
417 vnffgr
= self
._vnffgmgr
.create_vnffgr(vnffgr_request
,self
._vnffgd
_msg
.classifier
,vnffg_sff_list
)
418 except Exception as e
:
419 self
._log
.exception("VNFFG instantiation failed: %s", str(e
))
420 self
._vnffgr
_state
= VnffgRecordState
.FAILED
421 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFFGR %s failure" % (self
.id, vnffgr_request
.id))
423 self
._vnffgr
_state
= VnffgRecordState
.INSTANTIATION_PENDING
425 self
._log
.info("Instantiated VNFFGR :%s", vnffgr
)
426 self
._vnffgr
_state
= VnffgRecordState
.ACTIVE
428 self
._log
.info("Invoking update_state to update NSR state for NSR ID: %s", self
._nsr
.id)
429 yield from self
._nsr
.update_state()
431 def vnffgr_in_vnffgrm(self
):
432 """ Is there a VNFR record in VNFM """
433 if (self
._vnffgr
_state
== VnffgRecordState
.ACTIVE
or
434 self
._vnffgr
_state
== VnffgRecordState
.INSTANTIATION_PENDING
or
435 self
._vnffgr
_state
== VnffgRecordState
.FAILED
):
442 """ Terminate this VNFFGR """
443 if not self
.vnffgr_in_vnffgrm():
444 self
._log
.error("Ignoring terminate request for id %s in state %s",
445 self
.id, self
._vnffgr
_state
)
448 self
._log
.info("Terminating VNFFGR id:%s", self
.id)
449 self
._vnffgr
_state
= VnffgRecordState
.TERMINATE_PENDING
451 self
._vnffgmgr
.terminate_vnffgr(self
._vnffgr
_id
)
453 self
._vnffgr
_state
= VnffgRecordState
.TERMINATED
454 self
._log
.debug("Terminated VNFFGR id:%s", self
.id)
457 class VirtualLinkRecord(object):
458 """ Virtual Link Records class"""
459 XPATH
= "D,/vlr:vlr-catalog/vlr:vlr"
462 def create_record(dts
, log
, loop
, nsr_name
, vld_msg
, cloud_account_name
, om_datacenter
, ip_profile
, nsr_id
, restart_mode
=False):
463 """Creates a new VLR object based on the given data.
465 If restart mode is enabled, then we look for existing records in the
466 DTS and create a VLR records using the exiting data(ID)
471 vlr_obj
= VirtualLinkRecord(
484 res_iter
= yield from dts
.query_read(
485 "D,/vlr:vlr-catalog/vlr:vlr",
486 rwdts
.XactFlag
.MERGE
)
489 response
= yield from fut
490 vlr
= response
.result
492 # Check if the record is already present, if so use the ID of
493 # the existing record. Since the name of the record is uniquely
494 # formed we can use it as a search key!
495 if vlr
.name
== vlr_obj
.name
:
496 vlr_obj
.reset_id(vlr
.id)
501 def __init__(self
, dts
, log
, loop
, nsr_name
, vld_msg
, cloud_account_name
, om_datacenter
, ip_profile
, nsr_id
):
505 self
._nsr
_name
= nsr_name
506 self
._vld
_msg
= vld_msg
507 self
._cloud
_account
_name
= cloud_account_name
508 self
._om
_datacenter
_name
= om_datacenter
509 self
._assigned
_subnet
= None
510 self
._nsr
_id
= nsr_id
511 self
._ip
_profile
= ip_profile
512 self
._vlr
_id
= str(uuid
.uuid4())
513 self
._state
= VlRecordState
.INIT
514 self
._prev
_state
= None
515 self
._create
_time
= int(time
.time())
519 """ path for this object """
520 return "D,/vlr:vlr-catalog/vlr:vlr[vlr:id = '{}']".format(self
._vlr
_id
)
529 """ Get NSR name for this VL """
534 """ Virtual Link Desciptor """
538 def assigned_subnet(self
):
539 """ Subnet assigned to this VL"""
540 return self
._assigned
_subnet
545 Get the name for this VLR.
546 VLR name is "nsr name:VLD name"
548 if self
.vld_msg
.vim_network_name
:
549 return self
.vld_msg
.vim_network_name
550 elif self
.vld_msg
.name
== "multisite":
551 # This is a temporary hack to identify manually provisioned inter-site network
552 return self
.vld_msg
.name
554 return self
._nsr
_name
+ "." + self
.vld_msg
.name
557 def cloud_account_name(self
):
558 """ Cloud account that this VLR should be created in """
559 return self
._cloud
_account
_name
562 def om_datacenter_name(self
):
563 """ Datacenter that this VLR should be created in """
564 return self
._om
_datacenter
_name
568 """ Get the VLR path from VLR """
569 return (VirtualLinkRecord
.XPATH
+ "[vlr:id = '{}']").format(vlr
.id)
577 def state(self
, value
):
578 """ VLR set state """
582 def prev_state(self
):
583 """ VLR previous state """
584 return self
._prev
_state
587 def prev_state(self
, value
):
588 """ VLR set previous state """
589 self
._prev
_state
= value
593 """ Virtual Link Record message for Creating VLR in VNS """
594 vld_fields
= ["short_name",
602 vld_copy_dict
= {k
: v
for k
, v
in self
.vld_msg
.as_dict().items()
605 vlr_dict
= {"id": self
._vlr
_id
,
606 "nsr_id_ref": self
._nsr
_id
,
607 "vld_ref": self
.vld_msg
.id,
609 "create_time": self
._create
_time
,
610 "cloud_account": self
.cloud_account_name
,
611 "om_datacenter": self
.om_datacenter_name
,
614 if self
._ip
_profile
and self
._ip
_profile
.has_field('ip_profile_params'):
615 vlr_dict
['ip_profile_params' ] = self
._ip
_profile
.ip_profile_params
.as_dict()
617 vlr_dict
.update(vld_copy_dict
)
618 vlr
= RwVlrYang
.YangData_Vlr_VlrCatalog_Vlr
.from_dict(vlr_dict
)
621 def reset_id(self
, vlr_id
):
622 self
._vlr
_id
= vlr_id
624 def create_nsr_vlr_msg(self
, vnfrs
):
625 """ The VLR message"""
626 nsr_vlr
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vlr()
627 nsr_vlr
.vlr_ref
= self
._vlr
_id
628 nsr_vlr
.assigned_subnet
= self
.assigned_subnet
629 nsr_vlr
.cloud_account
= self
.cloud_account_name
630 nsr_vlr
.om_datacenter
= self
.om_datacenter_name
632 for conn
in self
.vld_msg
.vnfd_connection_point_ref
:
634 if (vnfr
.vnfd
.id == conn
.vnfd_id_ref
and
635 vnfr
.member_vnf_index
== conn
.member_vnf_index_ref
and
636 self
.cloud_account_name
== vnfr
.cloud_account_name
and
637 self
.om_datacenter_name
== vnfr
.om_datacenter_name
):
638 cp_entry
= nsr_vlr
.vnfr_connection_point_ref
.add()
639 cp_entry
.vnfr_id
= vnfr
.id
640 cp_entry
.connection_point
= conn
.vnfd_connection_point_ref
645 def instantiate(self
):
646 """ Instantiate this VL """
647 self
._log
.debug("Instaniating VLR key %s, vld %s",
648 self
.xpath
, self
._vld
_msg
)
650 self
._state
= VlRecordState
.INSTANTIATION_PENDING
651 self
._log
.debug("Executing VL create path:%s msg:%s",
652 self
.xpath
, self
.vlr_msg
)
654 with self
._dts
.transaction(flags
=0) as xact
:
655 block
= xact
.block_create()
656 block
.add_query_create(self
.xpath
, self
.vlr_msg
)
657 self
._log
.debug("Executing VL create path:%s msg:%s",
658 self
.xpath
, self
.vlr_msg
)
659 res_iter
= yield from block
.execute(now
=True)
665 self
._state
= VlRecordState
.FAILED
666 raise NsrInstantiationFailed("Failed NS %s instantiation due to empty response" % self
.id)
668 if vlr
.operational_status
== 'failed':
669 self
._log
.debug("NS Id:%s VL creation failed for vlr id %s", self
.id, vlr
.id)
670 self
._state
= VlRecordState
.FAILED
671 raise NsrInstantiationFailed("Failed VL %s instantiation (%s)" % (vlr
.id, vlr
.operational_status_details
))
673 self
._log
.info("Instantiated VL with xpath %s and vlr:%s",
675 self
._state
= VlRecordState
.ACTIVE
676 self
._assigned
_subnet
= vlr
.assigned_subnet
678 def vlr_in_vns(self
):
679 """ Is there a VLR record in VNS """
680 if (self
._state
== VlRecordState
.ACTIVE
or
681 self
._state
== VlRecordState
.INSTANTIATION_PENDING
or
682 self
._state
== VlRecordState
.TERMINATE_PENDING
or
683 self
._state
== VlRecordState
.FAILED
):
690 """ Terminate this VL """
691 if not self
.vlr_in_vns():
692 self
._log
.debug("Ignoring terminate request for id %s in state %s",
693 self
.id, self
._state
)
696 self
._log
.debug("Terminating VL id:%s", self
.id)
697 self
._state
= VlRecordState
.TERMINATE_PENDING
699 with self
._dts
.transaction(flags
=0) as xact
:
700 block
= xact
.block_create()
701 block
.add_query_delete(self
.xpath
)
702 yield from block
.execute(flags
=0, now
=True)
704 self
._state
= VlRecordState
.TERMINATED
705 self
._log
.debug("Terminated VL id:%s", self
.id)
708 class VnfRecordState(Enum
):
709 """ Vnf Record State """
711 INSTANTIATION_PENDING
= 102
713 TERMINATE_PENDING
= 104
718 class VirtualNetworkFunctionRecord(object):
719 """ Virtual Network Function Record class"""
720 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
724 def create_record(dts
, log
, loop
, vnfd
, const_vnfd_msg
, nsd_id
, nsr_name
,
725 cloud_account_name
, om_datacenter_name
, nsr_id
, group_name
, group_instance_id
,
726 placement_groups
, restart_mode
=False):
727 """Creates a new VNFR object based on the given data.
729 If restart mode is enabled, then we look for existing records in the
730 DTS and create a VNFR records using the exiting data(ID)
733 VirtualNetworkFunctionRecord
735 vnfr_obj
= VirtualNetworkFunctionRecord(
749 restart_mode
=restart_mode
)
752 res_iter
= yield from dts
.query_read(
753 "D,/vnfr:vnfr-catalog/vnfr:vnfr",
754 rwdts
.XactFlag
.MERGE
)
757 response
= yield from fut
758 vnfr
= response
.result
760 if vnfr
.name
== vnfr_obj
.name
:
761 vnfr_obj
.reset_id(vnfr
.id)
778 group_instance_id
=None,
779 placement_groups
= [],
780 restart_mode
= False):
785 self
._const
_vnfd
_msg
= const_vnfd_msg
786 self
._nsd
_id
= nsd_id
787 self
._nsr
_name
= nsr_name
788 self
._nsr
_id
= nsr_id
789 self
._cloud
_account
_name
= cloud_account_name
790 self
._om
_datacenter
_name
= om_datacenter_name
791 self
._group
_name
= group_name
792 self
._group
_instance
_id
= group_instance_id
793 self
._placement
_groups
= placement_groups
794 self
._config
_status
= NsrYang
.ConfigStates
.INIT
795 self
._create
_time
= int(time
.time())
797 self
._prev
_state
= VnfRecordState
.INIT
798 self
._state
= VnfRecordState
.INIT
799 self
._state
_failed
_reason
= None
801 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
804 self
._vnfr
_id
= str(uuid
.uuid4())
806 self
._vnfr
_msg
= self
.create_vnfr_msg()
807 self
._log
.debug("Set VNFR {} config type to {}".
808 format(self
.name
, self
.config_type
))
809 self
.restart_mode
= restart_mode
812 if group_name
is None and group_instance_id
is not None:
813 raise ValueError("Group instance id must not be provided with an empty group name")
823 return "D,/vnfr:vnfr-catalog/vnfr:vnfr[vnfr:id = '{}']".format(self
.id)
828 return self
._vnfr
_msg
831 def const_vnfr_msg(self
):
833 return RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_ConstituentVnfrRef(vnfr_id
=self
.id,cloud_account
=self
.cloud_account_name
,om_datacenter
=self
._om
_datacenter
_name
)
841 def cloud_account_name(self
):
842 """ Cloud account that this VNF should be created in """
843 return self
._cloud
_account
_name
846 def om_datacenter_name(self
):
847 """ Datacenter that this VNF should be created in """
848 return self
._om
_datacenter
_name
853 """ Is this VNF actve """
854 return True if self
._state
== VnfRecordState
.ACTIVE
else False
858 """ state of this VNF """
862 def state_failed_reason(self
):
863 """ Error message in case this VNF is in failed state """
864 return self
._state
_failed
_reason
867 def member_vnf_index(self
):
868 """ Member VNF index """
869 return self
._const
_vnfd
_msg
.member_vnf_index
874 return self
._nsr
_name
878 """ Name of this VNFR """
879 if self
._name
is not None:
882 name_tags
= [self
._nsr
_name
]
884 if self
._group
_name
is not None:
885 name_tags
.append(self
._group
_name
)
887 if self
._group
_instance
_id
is not None:
888 name_tags
.append(str(self
._group
_instance
_id
))
890 name_tags
.extend([self
.vnfd
.name
, str(self
.member_vnf_index
)])
892 self
._name
= "__".join(name_tags
)
897 def vnfr_xpath(vnfr
):
898 """ Get the VNFR path from VNFR """
899 return (VirtualNetworkFunctionRecord
.XPATH
+ "[vnfr:id = '{}']").format(vnfr
.id)
902 def config_type(self
):
903 cfg_types
= ['netconf', 'juju', 'script']
904 for method
in cfg_types
:
905 if self
._vnfd
.vnf_configuration
.has_field(method
):
910 def config_status(self
):
911 """Return the config status as YANG ENUM string"""
912 self
._log
.debug("Map VNFR {} config status {} ({})".
913 format(self
.name
, self
._config
_status
, self
.config_type
))
914 if self
.config_type
== 'none':
915 return 'config_not_needed'
916 elif self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
918 elif self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
923 def set_state(self
, state
):
924 """ set the state of this object """
925 self
._prev
_state
= self
._state
928 def reset_id(self
, vnfr_id
):
929 self
._vnfr
_id
= vnfr_id
930 self
._vnfr
_msg
= self
.create_vnfr_msg()
933 self
.config_store
.merge_vnfd_config(
936 self
.member_vnf_index
,
939 def create_vnfr_msg(self
):
940 """ VNFR message for this VNFR """
948 vnfd_copy_dict
= {k
: v
for k
, v
in self
._vnfd
.as_dict().items() if k
in vnfd_fields
}
951 "nsr_id_ref": self
._nsr
_id
,
953 "cloud_account": self
._cloud
_account
_name
,
954 "om_datacenter": self
._om
_datacenter
_name
,
955 "config_status": self
.config_status
957 vnfr_dict
.update(vnfd_copy_dict
)
959 vnfr
= RwVnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr
.from_dict(vnfr_dict
)
960 vnfr
.vnfd
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_Vnfd
.from_dict(self
.vnfd
.as_dict(),
961 ignore_missing_keys
=True)
962 vnfr
.member_vnf_index_ref
= self
.member_vnf_index
963 vnfr
.vnf_configuration
.from_dict(self
._vnfd
.vnf_configuration
.as_dict())
965 if self
._vnfd
.mgmt_interface
.has_field("port"):
966 vnfr
.mgmt_interface
.port
= self
._vnfd
.mgmt_interface
.port
968 for group_info
in self
._placement
_groups
:
969 group
= vnfr
.placement_groups_info
.add()
970 group
.from_dict(group_info
.as_dict())
972 # UI expects the monitoring param field to exist
973 vnfr
.monitoring_param
= []
975 self
._log
.debug("Get vnfr_msg for VNFR {} : {}".format(self
.name
, vnfr
))
979 def update_vnfm(self
):
980 self
._log
.debug("Send an update to VNFM for VNFR {} with {}".
981 format(self
.name
, self
.vnfr_msg
))
982 yield from self
._dts
.query_update(
984 rwdts
.XactFlag
.TRACE
,
988 def get_config_status(self
):
989 """Return the config status as YANG ENUM"""
990 return self
._config
_status
993 def set_config_status(self
, status
):
995 def status_to_string(status
):
997 NsrYang
.ConfigStates
.INIT
: 'init',
998 NsrYang
.ConfigStates
.CONFIGURING
: 'configuring',
999 NsrYang
.ConfigStates
.CONFIG_NOT_NEEDED
: 'config_not_needed',
1000 NsrYang
.ConfigStates
.CONFIGURED
: 'configured',
1001 NsrYang
.ConfigStates
.FAILED
: 'failed',
1004 return status_dc
[status
]
1006 self
._log
.debug("Update VNFR {} from {} ({}) to {}".
1007 format(self
.name
, self
._config
_status
,
1008 self
.config_type
, status
))
1009 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1010 self
._log
.error("Updating already configured VNFR {}".
1014 if self
._config
_status
!= status
:
1016 self
._config
_status
= status
1017 # I don't think this is used. Original implementor can check.
1018 # Caused Exception, so corrected it by status_to_string
1019 # But not sure whats the use of this variable?
1020 self
.vnfr_msg
.config_status
= status_to_string(status
)
1021 except Exception as e
:
1022 self
._log
.error("Exception=%s", str(e
))
1025 self
._log
.debug("Updated VNFR {} status to {}".format(self
.name
, status
))
1027 if self
._config
_status
!= NsrYang
.ConfigStates
.INIT
:
1029 # Publish only after VNFM has the VNFR created
1030 yield from self
.update_vnfm()
1031 except Exception as e
:
1032 self
._log
.error("Exception updating VNFM with new status {} of VNFR {}: {}".
1033 format(status
, self
.name
, e
))
1034 self
._log
.exception(e
)
1036 def is_configured(self
):
1037 if self
.config_type
== 'none':
1040 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1046 def instantiate(self
, nsr
):
1047 """ Instantiate this VNFR"""
1049 self
._log
.debug("Instaniating VNFR key %s, vnfd %s",
1050 self
.xpath
, self
._vnfd
)
1052 self
._log
.debug("Create VNF with xpath %s and vnfr %s",
1053 self
.xpath
, self
.vnfr_msg
)
1055 self
.set_state(VnfRecordState
.INSTANTIATION_PENDING
)
1057 def find_vlr_for_cp(conn
):
1058 """ Find VLR for the given connection point """
1059 for vlr
in nsr
.vlrs
:
1060 for vnfd_cp
in vlr
.vld_msg
.vnfd_connection_point_ref
:
1061 if (vnfd_cp
.vnfd_id_ref
== self
._vnfd
.id and
1062 vnfd_cp
.vnfd_connection_point_ref
== conn
.name
and
1063 vnfd_cp
.member_vnf_index_ref
== self
.member_vnf_index
and
1064 vlr
.cloud_account_name
== self
.cloud_account_name
):
1065 self
._log
.debug("Found VLR for cp_name:%s and vnf-index:%d",
1066 conn
.name
, self
.member_vnf_index
)
1070 # For every connection point in the VNFD fill in the identifier
1071 for conn_p
in self
._vnfd
.connection_point
:
1072 cpr
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_ConnectionPoint()
1073 cpr
.name
= conn_p
.name
1074 cpr
.type_yang
= conn_p
.type_yang
1075 if conn_p
.has_field('port_security_enabled'):
1076 cpr
.port_security_enabled
= conn_p
.port_security_enabled
1078 vlr_ref
= find_vlr_for_cp(conn_p
)
1080 msg
= "Failed to find VLR for cp = %s" % conn_p
.name
1081 self
._log
.debug("%s", msg
)
1082 # raise VirtualNetworkFunctionRecordError(msg)
1085 cpr
.vlr_ref
= vlr_ref
.id
1086 self
.vnfr_msg
.connection_point
.append(cpr
)
1087 self
._log
.debug("Connection point [%s] added, vnf id=%s vnfd id=%s",
1088 cpr
, self
.vnfr_msg
.id, self
.vnfr_msg
.vnfd
.id)
1090 if not self
.restart_mode
:
1091 yield from self
._dts
.query_create(self
.xpath
,
1095 yield from self
._dts
.query_update(self
.xpath
,
1099 self
._log
.info("Created VNF with xpath %s and vnfr %s",
1100 self
.xpath
, self
.vnfr_msg
)
1102 self
._log
.info("Instantiated VNFR with xpath %s and vnfd %s, vnfr %s",
1103 self
.xpath
, self
._vnfd
, self
.vnfr_msg
)
1106 def update_state(self
, vnfr_msg
):
1107 """ Update this VNFR"""
1108 if vnfr_msg
.operational_status
== "running":
1109 if self
.vnfr_msg
.operational_status
!= "running":
1110 yield from self
.is_active()
1111 elif vnfr_msg
.operational_status
== "failed":
1112 yield from self
.instantiation_failed(failed_reason
=vnfr_msg
.operational_status_details
)
1115 def is_active(self
):
1116 """ This VNFR is active """
1117 self
._log
.debug("VNFR %s is active", self
._vnfr
_id
)
1118 self
.set_state(VnfRecordState
.ACTIVE
)
1121 def instantiation_failed(self
, failed_reason
=None):
1122 """ This VNFR instantiation failed"""
1123 self
._log
.error("VNFR %s instantiation failed", self
._vnfr
_id
)
1124 self
.set_state(VnfRecordState
.FAILED
)
1125 self
._state
_failed
_reason
= failed_reason
1127 def vnfr_in_vnfm(self
):
1128 """ Is there a VNFR record in VNFM """
1129 if (self
._state
== VnfRecordState
.ACTIVE
or
1130 self
._state
== VnfRecordState
.INSTANTIATION_PENDING
or
1131 self
._state
== VnfRecordState
.FAILED
):
1137 def terminate(self
):
1138 """ Terminate this VNF """
1139 if not self
.vnfr_in_vnfm():
1140 self
._log
.debug("Ignoring terminate request for id %s in state %s",
1141 self
.id, self
._state
)
1144 self
._log
.debug("Terminating VNF id:%s", self
.id)
1145 self
.set_state(VnfRecordState
.TERMINATE_PENDING
)
1146 with self
._dts
.transaction(flags
=0) as xact
:
1147 block
= xact
.block_create()
1148 block
.add_query_delete(self
.xpath
)
1149 yield from block
.execute(flags
=0)
1150 self
.set_state(VnfRecordState
.TERMINATED
)
1151 self
._log
.debug("Terminated VNF id:%s", self
.id)
1154 class NetworkServiceStatus(object):
1155 """ A class representing the Network service's status """
1156 MAX_EVENTS_RECORDED
= 10
1157 """ Network service Status class"""
1158 def __init__(self
, dts
, log
, loop
):
1163 self
._state
= NetworkServiceRecordState
.INIT
1164 self
._events
= deque([])
1167 def create_notification(self
, evt
, evt_desc
, evt_details
):
1168 xp
= "N,/rw-nsr:nsm-notification"
1169 notif
= RwNsrYang
.YangNotif_RwNsr_NsmNotification()
1171 notif
.description
= evt_desc
1172 notif
.details
= evt_details
if evt_details
is not None else None
1174 yield from self
._dts
.query_create(xp
, rwdts
.XactFlag
.ADVISE
, notif
)
1175 self
._log
.info("Notification called by creating dts query: %s", notif
)
1177 def record_event(self
, evt
, evt_desc
, evt_details
):
1178 """ Record an event """
1179 self
._log
.debug("Recording event - evt %s, evt_descr %s len = %s",
1180 evt
, evt_desc
, len(self
._events
))
1181 if len(self
._events
) >= NetworkServiceStatus
.MAX_EVENTS_RECORDED
:
1182 self
._events
.popleft()
1183 self
._events
.append((int(time
.time()), evt
, evt_desc
,
1184 evt_details
if evt_details
is not None else None))
1186 self
._loop
.create_task(self
.create_notification(evt
,evt_desc
,evt_details
))
1188 def set_state(self
, state
):
1189 """ set the state of this status object """
1193 """ Return the state as a yang enum string """
1194 state_to_str_map
= {"INIT": "init",
1195 "VL_INIT_PHASE": "vl_init_phase",
1196 "VNF_INIT_PHASE": "vnf_init_phase",
1197 "VNFFG_INIT_PHASE": "vnffg_init_phase",
1198 "SCALING_GROUP_INIT_PHASE": "scaling_group_init_phase",
1199 "RUNNING": "running",
1200 "SCALING_OUT": "scaling_out",
1201 "SCALING_IN": "scaling_in",
1202 "TERMINATE_RCVD": "terminate_rcvd",
1203 "TERMINATE": "terminate",
1204 "VL_TERMINATE_PHASE": "vl_terminate_phase",
1205 "VNF_TERMINATE_PHASE": "vnf_terminate_phase",
1206 "VNFFG_TERMINATE_PHASE": "vnffg_terminate_phase",
1207 "TERMINATED": "terminated",
1209 "VL_INSTANTIATE": "vl_instantiate",
1210 "VL_TERMINATE": "vl_terminate",
1212 return state_to_str_map
[self
._state
.name
]
1216 """ State of this status object """
1221 """ Network Service Record as a message"""
1224 for entry
in self
._events
:
1225 event
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_OperationalEvents()
1228 event
.timestamp
, event
.event
, event
.description
, event
.details
= entry
1229 event_list
.append(event
)
1233 class NetworkServiceRecord(object):
1234 """ Network service record """
1235 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
1237 def __init__(self
, dts
, log
, loop
, nsm
, nsm_plugin
, nsr_cfg_msg
, sdn_account_name
, key_pairs
, restart_mode
=False,
1243 self
._nsr
_cfg
_msg
= nsr_cfg_msg
1244 self
._nsm
_plugin
= nsm_plugin
1245 self
._sdn
_account
_name
= sdn_account_name
1246 self
._vlr
_handler
= vlr_handler
1249 self
._nsr
_msg
= None
1250 self
._nsr
_regh
= None
1251 self
._key
_pairs
= key_pairs
1256 self
._param
_pools
= {}
1257 self
._scaling
_groups
= {}
1258 self
._create
_time
= int(time
.time())
1259 self
._op
_status
= NetworkServiceStatus(dts
, log
, loop
)
1260 self
._config
_status
= NsrYang
.ConfigStates
.CONFIGURING
1261 self
._config
_status
_details
= None
1263 self
.restart_mode
= restart_mode
1264 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
1265 self
._debug
_running
= False
1266 self
._is
_active
= False
1267 self
._vl
_phase
_completed
= False
1268 self
._vnf
_phase
_completed
= False
1269 self
.vlr_uptime_tasks
= {}
1272 # Initalise the state to init
1273 # The NSR moves through the following transitions
1274 # 1. INIT -> VLS_READY once all the VLs in the NSD are created
1275 # 2. VLS_READY - VNFS_READY when all the VNFs in the NSD are created
1276 # 3. VNFS_READY - READY when the NSR is published
1278 self
.set_state(NetworkServiceRecordState
.INIT
)
1280 self
.substitute_input_parameters
= InputParameterSubstitution(self
._log
)
1283 def nsm_plugin(self
):
1285 return self
._nsm
_plugin
1287 def set_state(self
, state
):
1288 """ Set state for this NSR"""
1289 self
._log
.debug("Setting state to %s", state
)
1290 # We are in init phase and is moving to the next state
1291 # The new state could be a FAILED state or VNF_INIIT_PHASE
1292 if self
.state
== NetworkServiceRecordState
.VL_INIT_PHASE
:
1293 self
._vl
_phase
_completed
= True
1295 if self
.state
== NetworkServiceRecordState
.VNF_INIT_PHASE
:
1296 self
._vnf
_phase
_completed
= True
1298 self
._op
_status
.set_state(state
)
1302 """ Get id for this NSR"""
1303 return self
._nsr
_cfg
_msg
.id
1307 """ Name of this network service record """
1308 return self
._nsr
_cfg
_msg
.name
1311 def cloud_account_name(self
):
1312 return self
._nsr
_cfg
_msg
.cloud_account
1315 def om_datacenter_name(self
):
1316 if self
._nsr
_cfg
_msg
.has_field('om_datacenter'):
1317 return self
._nsr
_cfg
_msg
.om_datacenter
1322 """State of this NetworkServiceRecord"""
1323 return self
._op
_status
.state
1327 """ Is this NSR active ?"""
1328 return True if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
else False
1332 """ VLRs associated with this NSR"""
1337 """ VNFRs associated with this NSR"""
1342 """ VNFFGRs associated with this NSR"""
1343 return self
._vnffgrs
1346 def scaling_groups(self
):
1347 """ Scaling groups associated with this NSR """
1348 return self
._scaling
_groups
1351 def param_pools(self
):
1352 """ Parameter value pools associated with this NSR"""
1353 return self
._param
_pools
1356 def nsr_cfg_msg(self
):
1357 return self
._nsr
_cfg
_msg
1360 def nsr_cfg_msg(self
, msg
):
1361 self
._nsr
_cfg
_msg
= msg
1365 """ NSD Protobuf for this NSR """
1366 if self
._nsd
is not None:
1368 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
1373 """ NSD ID for this NSR """
1374 return self
.nsd_msg
.id
1378 ''' Get a new job id for config primitive'''
1383 def config_status(self
):
1384 """ Config status for NSR """
1385 return self
._config
_status
1387 def resolve_placement_group_cloud_construct(self
, input_group
):
1389 Returns the cloud specific construct for placement group
1391 copy_dict
= ['name', 'requirement', 'strategy']
1393 for group_info
in self
._nsr
_cfg
_msg
.nsd_placement_group_maps
:
1394 if group_info
.placement_group_ref
== input_group
.name
:
1395 group
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_PlacementGroupsInfo()
1396 group_dict
= {k
:v
for k
,v
in
1397 group_info
.as_dict().items() if k
!= 'placement_group_ref'}
1398 for param
in copy_dict
:
1399 group_dict
.update({param
: getattr(input_group
, param
)})
1400 group
.from_dict(group_dict
)
1406 return "NSR(name={}, nsd_id={}, cloud_account={})".format(
1407 self
.name
, self
.nsd_id
, self
.cloud_account_name
1410 def _get_vnfd(self
, vnfd_id
, config_xact
):
1411 """ Fetch vnfd msg for the passed vnfd id """
1412 return self
._nsm
.get_vnfd(vnfd_id
, config_xact
)
1414 def _get_vnfd_cloud_account(self
, vnfd_member_index
):
1415 """ Fetch Cloud Account for the passed vnfd id """
1416 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1417 vim_accounts
= [(vnf
.cloud_account
,vnf
.om_datacenter
) for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map \
1418 if vnfd_member_index
== vnf
.member_vnf_index_ref
]
1419 if vim_accounts
and vim_accounts
[0]:
1420 return vim_accounts
[0]
1421 return (self
.cloud_account_name
,self
.om_datacenter_name
)
1423 def _get_constituent_vnfd_msg(self
, vnf_index
):
1424 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1425 if const_vnfd
.member_vnf_index
== vnf_index
:
1428 raise ValueError("Constituent VNF index %s not found" % vnf_index
)
1430 def record_event(self
, evt
, evt_desc
, evt_details
=None, state
=None):
1431 """ Record an event """
1432 self
._op
_status
.record_event(evt
, evt_desc
, evt_details
)
1433 if state
is not None:
1434 self
.set_state(state
)
1436 def scaling_trigger_str(self
, trigger
):
1437 SCALING_TRIGGER_STRS
= {
1438 NsdYang
.ScalingTrigger
.PRE_SCALE_IN
: 'pre-scale-in',
1439 NsdYang
.ScalingTrigger
.POST_SCALE_IN
: 'post-scale-in',
1440 NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
: 'pre-scale-out',
1441 NsdYang
.ScalingTrigger
.POST_SCALE_OUT
: 'post-scale-out',
1444 return SCALING_TRIGGER_STRS
[trigger
]
1445 except Exception as e
:
1446 self
._log
.error("Scaling trigger mapping error for {} : {}".
1448 self
._log
.exception(e
)
1449 return "Unknown trigger"
1452 def instantiate_vls(self
):
1454 This function instantiates VLs for every VL in this Network Service
1456 self
._log
.debug("Instantiating %d VLs in NSD id %s", len(self
._vlrs
),
1458 for vlr
in self
._vlrs
:
1459 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1460 vlr
.state
= VlRecordState
.ACTIVE
1461 self
.vlr_uptime_tasks
[vlr
.id] = self
._loop
.create_task(self
.vlr_uptime_update(vlr
))
1464 def vlr_uptime_update(self
, vlr
):
1467 vlr_
= RwVlrYang
.YangData_Vlr_VlrCatalog_Vlr
.from_dict({'id': vlr
.id})
1469 vlr_
.uptime
= int(time
.time()) - vlr
._create
_time
1470 yield from self
._vlr
_handler
.update(None, VirtualLinkRecord
.vlr_xpath(vlr
), vlr_
)
1471 yield from asyncio
.sleep(2, loop
=self
._loop
)
1472 except asyncio
.CancelledError
:
1473 self
._log
.debug("Received cancellation request for vlr_uptime_update task")
1474 yield from self
._vlr
_handler
.delete(None, VirtualLinkRecord
.vlr_xpath(vlr
))
1478 def create(self
, config_xact
):
1479 """ Create this network service"""
1480 # Create virtual links for all the external vnf
1481 # connection points in this NS
1482 yield from self
.create_vls()
1484 # Create VNFs in this network service
1485 yield from self
.create_vnfs(config_xact
)
1487 # Create VNFFG for network service
1488 self
.create_vnffgs()
1490 # Create Scaling Groups for each scaling group in NSD
1491 self
.create_scaling_groups()
1493 # Create Parameter Pools
1494 self
.create_param_pools()
1497 def apply_scale_group_config_script(self
, script
, group
, scale_instance
, trigger
, vnfrs
=None):
1498 """ Apply config based on script for scale group """
1501 def add_vnfrs_data(vnfrs_list
):
1502 """ Add as a dict each of the VNFRs data """
1504 for vnfr
in vnfrs_list
:
1505 self
._log
.debug("Add VNFR {} data".format(vnfr
))
1507 vnfr_data
['name'] = vnfr
.name
1508 if trigger
in [NsdYang
.ScalingTrigger
.PRE_SCALE_IN
, NsdYang
.ScalingTrigger
.POST_SCALE_OUT
]:
1509 # Get VNF management and other IPs, etc
1510 opdata
= yield from self
.fetch_vnfr(vnfr
.xpath
)
1511 self
._log
.debug("VNFR {} op data: {}".format(vnfr
.name
, opdata
))
1513 vnfr_data
['rw_mgmt_ip'] = opdata
.mgmt_interface
.ip_address
1514 vnfr_data
['rw_mgmt_port'] = opdata
.mgmt_interface
.port
1515 except Exception as e
:
1516 self
._log
.error("Unable to get management IP for vnfr {}:{}".
1517 format(vnfr
.name
, e
))
1520 vnfr_data
['connection_points'] = []
1521 for cp
in opdata
.connection_point
:
1523 con_pt
['name'] = cp
.name
1524 con_pt
['ip_address'] = cp
.ip_address
1525 vnfr_data
['connection_points'].append(con_pt
)
1526 except Exception as e
:
1527 self
._log
.error("Exception getting connections points for VNFR {}: {}".
1528 format(vnfr
.name
, e
))
1530 vnfrs_data
.append(vnfr_data
)
1531 self
._log
.debug("VNFRs data: {}".format(vnfrs_data
))
1535 def add_nsr_data(nsr
):
1537 nsr_data
['name'] = nsr
.name
1540 if script
is None or len(script
) == 0:
1541 self
._log
.error("Script not provided for scale group config: {}".format(group
.name
))
1544 if script
[0] == '/':
1547 path
= os
.path
.join(os
.environ
['RIFT_INSTALL'], "usr/bin", script
)
1548 if not os
.path
.exists(path
):
1549 self
._log
.error("Config faled for scale group {}: Script does not exist at {}".
1550 format(group
.name
, path
))
1553 # Build a YAML file with all parameters for the script to execute
1554 # The data consists of 5 sections
1556 # 2. Scale group config
1557 # 3. VNFRs in the scale group
1558 # 4. VNFRs outside scale group
1561 data
['trigger'] = group
.trigger_map(trigger
)
1562 data
['config'] = group
.group_msg
.as_dict()
1565 data
["vnfrs_in_group"] = yield from add_vnfrs_data(vnfrs
)
1567 data
["vnfrs_in_group"] = yield from add_vnfrs_data(scale_instance
.vnfrs
)
1569 data
["vnfrs_others"] = yield from add_vnfrs_data(self
.vnfrs
.values())
1570 data
["nsr"] = add_nsr_data(self
)
1573 with tempfile
.NamedTemporaryFile(delete
=False) as tmp_file
:
1574 tmp_file
.write(yaml
.dump(data
, default_flow_style
=True)
1577 self
._log
.debug("Creating a temp file: {} with input data: {}".
1578 format(tmp_file
.name
, data
))
1580 cmd
= "{} {}".format(path
, tmp_file
.name
)
1581 self
._log
.debug("Running the CMD: {}".format(cmd
))
1582 proc
= yield from asyncio
.create_subprocess_shell(cmd
, loop
=self
._loop
)
1583 rc
= yield from proc
.wait()
1585 self
._log
.error("The script {} for scale group {} config returned: {}".
1586 format(script
, group
.name
, rc
))
1594 def apply_scaling_group_config(self
, trigger
, group
, scale_instance
, vnfrs
=None):
1595 """ Apply the config for the scaling group based on trigger """
1596 if group
is None or scale_instance
is None:
1600 def update_config_status(success
=True, err_msg
=None):
1601 self
._log
.debug("Update %s config status to %r : %s",
1602 scale_instance
, success
, err_msg
)
1603 if (scale_instance
.config_status
== "failed"):
1604 # Do not update the config status if it is already in failed state
1607 if scale_instance
.config_status
== "configured":
1608 # Update only to failed state an already configured scale instance
1610 scale_instance
.config_status
= "failed"
1611 scale_instance
.config_err_msg
= err_msg
1612 yield from self
.update_state()
1614 # We are in configuring state
1615 # Only after post scale out mark instance as configured
1616 if trigger
== NsdYang
.ScalingTrigger
.POST_SCALE_OUT
:
1618 scale_instance
.config_status
= "configured"
1620 scale_instance
.config_status
= "failed"
1621 scale_instance
.config_err_msg
= err_msg
1622 yield from self
.update_state()
1624 config
= group
.trigger_config(trigger
)
1628 self
._log
.debug("Scaling group {} config: {}".format(group
.name
, config
))
1629 if config
.has_field("ns_config_primitive_name_ref"):
1630 config_name
= config
.ns_config_primitive_name_ref
1631 nsd_msg
= self
.nsd_msg
1632 config_primitive
= None
1633 for ns_cfg_prim
in nsd_msg
.service_primitive
:
1634 if ns_cfg_prim
.name
== config_name
:
1635 config_primitive
= ns_cfg_prim
1638 if config_primitive
is None:
1639 raise ValueError("Could not find ns_cfg_prim %s in nsr %s" % (config_name
, self
.name
))
1641 self
._log
.debug("Scaling group {} config primitive: {}".format(group
.name
, config_primitive
))
1642 if config_primitive
.has_field("user_defined_script"):
1643 rc
= yield from self
.apply_scale_group_config_script(config_primitive
.user_defined_script
,
1644 group
, scale_instance
, trigger
, vnfrs
)
1647 err_msg
= "Failed config for trigger {} using config script '{}'". \
1648 format(self
.scaling_trigger_str(trigger
),
1649 config_primitive
.user_defined_script
)
1650 yield from update_config_status(success
=rc
, err_msg
=err_msg
)
1653 err_msg
= "Failed config for trigger {} as config script is not specified". \
1654 format(self
.scaling_trigger_str(trigger
))
1655 yield from update_config_status(success
=False, err_msg
=err_msg
)
1656 raise NotImplementedError("Only script based config support for scale group for now: {}".
1659 err_msg
= "Failed config for trigger {} as config primitive is not specified".\
1660 format(self
.scaling_trigger_str(trigger
))
1661 yield from update_config_status(success
=False, err_msg
=err_msg
)
1662 self
._log
.error("Config primitive not specified for config action in scale group %s" %
1666 def create_scaling_groups(self
):
1667 """ This function creates a NSScalingGroup for every scaling
1668 group defined in he NSD"""
1670 for scaling_group_msg
in self
.nsd_msg
.scaling_group_descriptor
:
1671 self
._log
.debug("Found scaling_group %s in nsr id %s",
1672 scaling_group_msg
.name
, self
.id)
1674 group_record
= scale_group
.ScalingGroup(
1679 self
._scaling
_groups
[group_record
.name
] = group_record
1682 def create_scale_group_instance(self
, group_name
, index
, config_xact
, is_default
=False):
1683 group
= self
._scaling
_groups
[group_name
]
1684 scale_instance
= group
.create_instance(index
, is_default
)
1688 self
._log
.debug("Creating %u VNFs associated with NS id %s scaling group %s",
1689 len(self
.nsd_msg
.constituent_vnfd
), self
.id, self
)
1692 for vnf_index
, count
in group
.vnf_index_count_map
.items():
1693 const_vnfd_msg
= self
._get
_constituent
_vnfd
_msg
(vnf_index
)
1694 vnfd_msg
= self
._get
_vnfd
(const_vnfd_msg
.vnfd_id_ref
, config_xact
)
1696 cloud_account_name
, om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd_msg
.member_vnf_index
)
1697 if cloud_account_name
is None:
1698 cloud_account_name
= self
.cloud_account_name
1699 for _
in range(count
):
1700 vnfr
= yield from self
.create_vnf_record(vnfd_msg
, const_vnfd_msg
, cloud_account_name
, om_datacenter_name
, group_name
, index
)
1701 scale_instance
.add_vnfr(vnfr
)
1707 def instantiate_instance():
1708 self
._log
.debug("Creating %s VNFRS", scale_instance
)
1709 vnfrs
= yield from create_vnfs()
1710 yield from self
.publish()
1712 self
._log
.debug("Instantiating %s VNFRS for %s", len(vnfrs
), scale_instance
)
1713 scale_instance
.operational_status
= "vnf_init_phase"
1714 yield from self
.update_state()
1717 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
,
1718 group
, scale_instance
, vnfrs
)
1720 self
._log
.error("Pre scale out config for scale group {} ({}) failed".
1721 format(group
.name
, index
))
1722 scale_instance
.operational_status
= "failed"
1724 yield from self
.instantiate_vnfs(vnfrs
)
1726 except Exception as e
:
1727 self
._log
.exception("Failed to begin instantiatiation of vnfs for scale group {}: {}".
1728 format(group
.name
, e
))
1729 self
._log
.exception(e
)
1730 scale_instance
.operational_status
= "failed"
1732 yield from self
.update_state()
1734 yield from instantiate_instance()
1737 def delete_scale_group_instance(self
, group_name
, index
):
1738 group
= self
._scaling
_groups
[group_name
]
1739 scale_instance
= group
.get_instance(index
)
1740 if scale_instance
.is_default
:
1741 raise ScalingOperationError("Cannot terminate a default scaling group instance")
1743 scale_instance
.operational_status
= "terminate"
1744 yield from self
.update_state()
1747 def terminate_instance():
1748 self
._log
.debug("Terminating %s VNFRS" % scale_instance
)
1749 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_IN
,
1750 group
, scale_instance
)
1752 self
._log
.error("Pre scale in config for scale group {} ({}) failed".
1753 format(group
.name
, index
))
1755 # Going ahead with terminate, even if there is an error in pre-scale-in config
1756 # as this could be result of scale out failure and we need to cleanup this group
1757 yield from self
.terminate_vnfrs(scale_instance
.vnfrs
)
1758 group
.delete_instance(index
)
1760 scale_instance
.operational_status
= "vnf_terminate_phase"
1761 yield from self
.update_state()
1763 yield from terminate_instance()
1766 def _update_scale_group_instances_status(self
):
1768 def post_scale_out_task(group
, instance
):
1769 # Apply post scale out config once all VNFRs are active
1770 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_OUT
,
1772 instance
.operational_status
= "running"
1774 self
._log
.debug("Scale out for group {} and instance {} succeeded".
1775 format(group
.name
, instance
.instance_id
))
1777 self
._log
.error("Post scale out config for scale group {} ({}) failed".
1778 format(group
.name
, instance
.instance_id
))
1780 yield from self
.update_state()
1782 group_instances
= {group
: group
.instances
for group
in self
._scaling
_groups
.values()}
1783 for group
, instances
in group_instances
.items():
1784 self
._log
.debug("Updating %s instance status", group
)
1785 for instance
in instances
:
1786 instance_vnf_state_list
= [vnfr
.state
for vnfr
in instance
.vnfrs
]
1787 self
._log
.debug("Got vnfr instance states: %s", instance_vnf_state_list
)
1788 if instance
.operational_status
== "vnf_init_phase":
1789 if all([state
== VnfRecordState
.ACTIVE
for state
in instance_vnf_state_list
]):
1790 instance
.operational_status
= "running"
1792 # Create a task for post scale out to allow us to sleep before attempting
1793 # to configure newly created VM's
1794 self
._loop
.create_task(post_scale_out_task(group
, instance
))
1796 elif any([state
== VnfRecordState
.FAILED
for state
in instance_vnf_state_list
]):
1797 self
._log
.debug("Scale out for group {} and instance {} failed".
1798 format(group
.name
, instance
.instance_id
))
1799 instance
.operational_status
= "failed"
1801 elif instance
.operational_status
== "vnf_terminate_phase":
1802 if all([state
== VnfRecordState
.TERMINATED
for state
in instance_vnf_state_list
]):
1803 instance
.operational_status
= "terminated"
1804 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_IN
,
1807 self
._log
.debug("Scale in for group {} and instance {} succeeded".
1808 format(group
.name
, instance
.instance_id
))
1810 self
._log
.error("Post scale in config for scale group {} ({}) failed".
1811 format(group
.name
, instance
.instance_id
))
1813 def create_vnffgs(self
):
1814 """ This function creates VNFFGs for every VNFFG in the NSD
1815 associated with this NSR"""
1817 for vnffgd
in self
.nsd_msg
.vnffgd
:
1818 self
._log
.debug("Found vnffgd %s in nsr id %s", vnffgd
, self
.id)
1819 vnffgr
= VnffgRecord(self
._dts
,
1822 self
._nsm
._vnffgmgr
,
1826 self
._sdn
_account
_name
1828 self
._vnffgrs
[vnffgr
.id] = vnffgr
1830 def resolve_vld_ip_profile(self
, nsd_msg
, vld
):
1831 self
._log
.debug("Receieved ip profile ref is %s",vld
.ip_profile_ref
)
1832 if not vld
.has_field('ip_profile_ref'):
1834 profile
= [profile
for profile
in nsd_msg
.ip_profiles
if profile
.name
== vld
.ip_profile_ref
]
1835 return profile
[0] if profile
else None
1838 def _create_vls(self
, vld
, cloud_account
,om_datacenter
):
1839 """Create a VLR in the cloud account specified using the given VLD
1843 cloud_account : Cloud account name
1848 vlr
= yield from VirtualLinkRecord
.create_record(
1856 self
.resolve_vld_ip_profile(self
.nsd_msg
, vld
),
1858 restart_mode
=self
.restart_mode
)
1862 def _extract_cloud_accounts_for_vl(self
, vld
):
1864 Extracts the list of cloud accounts from the NS Config obj
1867 1. Cloud accounts based connection point (vnf_cloud_account_map)
1869 vld : VLD yang object
1874 cloud_account_list
= []
1876 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1877 # Handle case where cloud_account is None
1879 for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1880 if vnf
.cloud_account
is not None or vnf
.om_datacenter
is not None:
1881 vnf_cloud_map
[vnf
.member_vnf_index_ref
] = (vnf
.cloud_account
,vnf
.om_datacenter
)
1883 for vnfc
in vld
.vnfd_connection_point_ref
:
1884 cloud_account
= vnf_cloud_map
.get(
1885 vnfc
.member_vnf_index_ref
,
1886 (self
.cloud_account_name
,self
.om_datacenter_name
))
1888 cloud_account_list
.append(cloud_account
)
1890 if self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1891 for vld_map
in self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1892 if vld_map
.vld_id_ref
== vld
.id:
1893 for cloud_account
in vld_map
.cloud_accounts
:
1894 cloud_account_list
.extend((cloud_account
,None))
1895 for om_datacenter
in vld_map
.om_datacenters
:
1896 cloud_account_list
.extend((None,om_datacenter
))
1898 # If no config has been provided then fall-back to the default
1900 if not cloud_account_list
:
1901 cloud_account_list
= [(self
.cloud_account_name
,self
.om_datacenter_name
)]
1903 self
._log
.debug("VL {} cloud accounts: {}".
1904 format(vld
.name
, cloud_account_list
))
1905 return set(cloud_account_list
)
1908 def create_vls(self
):
1909 """ This function creates VLs for every VLD in the NSD
1910 associated with this NSR"""
1911 for vld
in self
.nsd_msg
.vld
:
1913 self
._log
.debug("Found vld %s in nsr id %s", vld
, self
.id)
1914 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1915 for cloud_account
,om_datacenter
in cloud_account_list
:
1916 vlr
= yield from self
._create
_vls
(vld
, cloud_account
,om_datacenter
)
1917 self
._vlrs
.append(vlr
)
1921 def create_vl_instance(self
, vld
):
1922 self
._log
.debug("Create VL for {}: {}".format(self
.id, vld
.as_dict()))
1923 # Check if the VL is already present
1925 for vl
in self
._vlrs
:
1926 if vl
.vld_msg
.id == vld
.id:
1927 self
._log
.debug("The VLD %s already in NSR %s as VLR %s with status %s",
1928 vld
.id, self
.id, vl
.id, vl
.state
)
1930 if vlr
.state
!= VlRecordState
.TERMINATED
:
1931 err_msg
= "VLR for VL %s in NSR %s already instantiated", \
1933 self
._log
.error(err_msg
)
1934 raise NsrVlUpdateError(err_msg
)
1938 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1939 for account
,om_datacenter
in cloud_account_list
:
1940 vlr
= yield from self
._create
_vls
(vld
, account
,om_datacenter
)
1941 self
._vlrs
.append(vlr
)
1943 vlr
.state
= VlRecordState
.INSTANTIATION_PENDING
1944 yield from self
.update_state()
1947 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1948 vlr
.state
= VlRecordState
.ACTIVE
1950 except Exception as e
:
1951 err_msg
= "Error instantiating VL for NSR {} and VLD {}: {}". \
1952 format(self
.id, vld
.id, e
)
1953 self
._log
.error(err_msg
)
1954 self
._log
.exception(e
)
1955 vlr
.state
= VlRecordState
.FAILED
1957 yield from self
.update_state()
1960 def delete_vl_instance(self
, vld
):
1961 for vlr
in self
._vlrs
:
1962 if vlr
.vld_msg
.id == vld
.id:
1963 self
._log
.debug("Found VLR %s for VLD %s in NSR %s",
1964 vlr
.id, vld
.id, self
.id)
1965 vlr
.state
= VlRecordState
.TERMINATE_PENDING
1966 yield from self
.update_state()
1969 yield from self
.nsm_plugin
.terminate_vl(vlr
)
1970 vlr
.state
= VlRecordState
.TERMINATED
1971 self
._vlrs
.remove(vlr
)
1973 except Exception as e
:
1974 err_msg
= "Error terminating VL for NSR {} and VLD {}: {}". \
1975 format(self
.id, vld
.id, e
)
1976 self
._log
.error(err_msg
)
1977 self
._log
.exception(e
)
1978 vlr
.state
= VlRecordState
.FAILED
1980 yield from self
.update_state()
1984 def create_vnfs(self
, config_xact
):
1986 This function creates VNFs for every VNF in the NSD
1987 associated with this NSR
1989 self
._log
.debug("Creating %u VNFs associated with this NS id %s",
1990 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
1992 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1993 if not const_vnfd
.start_by_default
:
1994 self
._log
.debug("start_by_default set to False in constituent VNF (%s). Skipping start.",
1995 const_vnfd
.member_vnf_index
)
1998 vnfd_msg
= self
._get
_vnfd
(const_vnfd
.vnfd_id_ref
, config_xact
)
1999 cloud_account_name
,om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd
.member_vnf_index
)
2000 if cloud_account_name
is None:
2001 cloud_account_name
= self
.cloud_account_name
2002 yield from self
.create_vnf_record(vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
)
2005 def get_placement_groups(self
, vnfd_msg
, const_vnfd
):
2006 placement_groups
= []
2007 for group
in self
.nsd_msg
.placement_groups
:
2008 for member_vnfd
in group
.member_vnfd
:
2009 if (member_vnfd
.vnfd_id_ref
== vnfd_msg
.id) and \
2010 (member_vnfd
.member_vnf_index_ref
== const_vnfd
.member_vnf_index
):
2011 group_info
= self
.resolve_placement_group_cloud_construct(group
)
2012 if group_info
is None:
2013 self
._log
.error("Could not resolve cloud-construct for placement group: %s", group
.name
)
2014 ### raise PlacementGroupError("Could not resolve cloud-construct for placement group: {}".format(group.name))
2016 self
._log
.info("Successfully resolved cloud construct for placement group: %s for VNF: %s (Member Index: %s)",
2019 const_vnfd
.member_vnf_index
)
2020 placement_groups
.append(group_info
)
2021 return placement_groups
2024 def create_vnf_record(self
, vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
, group_name
=None, group_instance_id
=None):
2025 # Fetch the VNFD associated with this VNF
2026 placement_groups
= self
.get_placement_groups(vnfd_msg
, const_vnfd
)
2027 self
._log
.info("Cloud Account for VNF %d is %s",const_vnfd
.member_vnf_index
,cloud_account_name
)
2028 self
._log
.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s",
2030 const_vnfd
.member_vnf_index
,
2031 [ group
.name
for group
in placement_groups
])
2032 vnfr
= yield from VirtualNetworkFunctionRecord
.create_record(self
._dts
,
2045 restart_mode
=self
.restart_mode
,
2047 if vnfr
.id in self
._vnfrs
:
2048 err
= "VNF with VNFR id %s already in vnf list" % (vnfr
.id,)
2049 raise NetworkServiceRecordError(err
)
2051 self
._vnfrs
[vnfr
.id] = vnfr
2052 self
._nsm
.vnfrs
[vnfr
.id] = vnfr
2054 yield from vnfr
.set_config_status(NsrYang
.ConfigStates
.INIT
)
2056 self
._log
.debug("Added VNFR %s to NSM VNFR list with id %s",
2062 def create_param_pools(self
):
2063 for param_pool
in self
.nsd_msg
.parameter_pool
:
2064 self
._log
.debug("Found parameter pool %s in nsr id %s", param_pool
, self
.id)
2066 start_value
= param_pool
.range.start_value
2067 end_value
= param_pool
.range.end_value
2068 if end_value
< start_value
:
2069 raise NetworkServiceRecordError(
2070 "Parameter pool %s has invalid range (start: {}, end: {})".format(
2071 start_value
, end_value
2075 self
._param
_pools
[param_pool
.name
] = config_value_pool
.ParameterValuePool(
2078 range(start_value
, end_value
)
2082 def fetch_vnfr(self
, vnfr_path
):
2083 """ Fetch VNFR record """
2085 self
._log
.debug("Fetching VNFR with key %s while instantiating %s",
2087 res_iter
= yield from self
._dts
.query_read(vnfr_path
, rwdts
.XactFlag
.MERGE
)
2089 for ent
in res_iter
:
2090 res
= yield from ent
2096 def instantiate_vnfs(self
, vnfrs
):
2098 This function instantiates VNFs for every VNF in this Network Service
2100 self
._log
.debug("Instantiating %u VNFs in NS %s", len(vnfrs
), self
.id)
2102 self
._log
.debug("Instantiating VNF: %s in NS %s", vnf
, self
.id)
2103 yield from self
.nsm_plugin
.instantiate_vnf(self
, vnf
)
2106 def instantiate_vnffgs(self
):
2108 This function instantiates VNFFGs for every VNFFG in this Network Service
2110 self
._log
.debug("Instantiating %u VNFFGs in NS %s",
2111 len(self
.nsd_msg
.vnffgd
), self
.id)
2112 for _
, vnfr
in self
.vnfrs
.items():
2113 while vnfr
.state
in [VnfRecordState
.INSTANTIATION_PENDING
, VnfRecordState
.INIT
]:
2114 self
._log
.debug("Received vnfr state for vnfr %s is %s; retrying",vnfr
.name
,vnfr
.state
)
2115 yield from asyncio
.sleep(2, loop
=self
._loop
)
2116 if vnfr
.state
== VnfRecordState
.ACTIVE
:
2117 self
._log
.debug("Received vnfr state for vnfr %s is %s ",vnfr
.name
,vnfr
.state
)
2120 self
._log
.debug("Received vnfr state for vnfr %s is %s; failing vnffg creation",vnfr
.name
,vnfr
.state
)
2121 self
._vnffgr
_state
= VnffgRecordState
.FAILED
2124 self
._log
.info("Waiting for 90 seconds for VMs to come up")
2125 yield from asyncio
.sleep(90, loop
=self
._loop
)
2126 self
._log
.info("Starting VNFFG orchestration")
2127 for vnffg
in self
._vnffgrs
.values():
2128 self
._log
.debug("Instantiating VNFFG: %s in NS %s", vnffg
, self
.id)
2129 yield from vnffg
.instantiate()
2132 def instantiate_scaling_instances(self
, config_xact
):
2133 """ Instantiate any default scaling instances in this Network Service """
2134 for group
in self
._scaling
_groups
.values():
2135 for i
in range(group
.min_instance_count
):
2136 self
._log
.debug("Instantiating %s default scaling instance %s", group
, i
)
2137 yield from self
.create_scale_group_instance(
2138 group
.name
, i
, config_xact
, is_default
=True
2141 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2142 if group_msg
.scaling_group_name_ref
!= group
.name
:
2145 for instance
in group_msg
.instance
:
2146 self
._log
.debug("Reloading %s scaling instance %s", group_msg
, instance
.id)
2147 yield from self
.create_scale_group_instance(
2148 group
.name
, instance
.id, config_xact
, is_default
=False
2151 def has_scaling_instances(self
):
2152 """ Return boolean indicating if the network service has default scaling groups """
2153 for group
in self
._scaling
_groups
.values():
2154 if group
.min_instance_count
> 0:
2157 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2158 if len(group_msg
.instance
) > 0:
2165 """ This function publishes this NSR """
2166 self
._nsr
_msg
= self
.create_msg()
2168 self
._log
.debug("Publishing the NSR with xpath %s and nsr %s",
2172 if self
._debug
_running
:
2173 self
._log
.debug("Publishing NSR in RUNNING state!")
2176 with self
._dts
.transaction() as xact
:
2177 yield from self
._nsm
.nsr_handler
.update(xact
, self
.nsr_xpath
, self
._nsr
_msg
)
2178 if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
:
2179 self
._debug
_running
= True
2182 def unpublish(self
, xact
):
2183 """ Unpublish this NSR object """
2184 self
._log
.debug("Unpublishing Network service id %s", self
.id)
2185 yield from self
._nsm
.nsr_handler
.delete(xact
, self
.nsr_xpath
)
2188 def nsr_xpath(self
):
2189 """ Returns the xpath associated with this NSR """
2191 "D,/nsr:ns-instance-opdata" +
2192 "/nsr:nsr[nsr:ns-instance-config-ref = '{}']"
2196 def xpath_from_nsr(nsr
):
2197 """ Returns the xpath associated with this NSR op data"""
2198 return (NetworkServiceRecord
.XPATH
+
2199 "[nsr:ns-instance-config-ref = '{}']").format(nsr
.id)
2202 def nsd_xpath(self
):
2203 """ Return NSD config xpath."""
2205 "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}']"
2206 ).format(self
.nsd_id
)
2209 def instantiate(self
, config_xact
):
2210 """"Instantiates a NetworkServiceRecord.
2212 This function instantiates a Network service
2213 which involves the following steps,
2215 * Instantiate every VL in NSD by sending create VLR request to DTS.
2216 * Instantiate every VNF in NSD by sending create VNF reuqest to DTS.
2217 * Publish the NSR details to DTS
2220 nsr: The NSR configuration request containing nsr-id and nsd
2221 config_xact: The configuration transaction which initiated the instatiation
2224 NetworkServiceRecordError if the NSR creation fails
2230 self
._log
.debug("Instantiating NS - %s xact - %s", self
, config_xact
)
2232 # Move the state to INIITALIZING
2233 self
.set_state(NetworkServiceRecordState
.INIT
)
2235 event_descr
= "Instantiation Request Received NSR Id:%s" % self
.id
2236 self
.record_event("instantiating", event_descr
)
2239 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
2242 # Update ref count if nsd present in catalog
2243 self
._nsm
.get_nsd_ref(self
.nsd_id
)
2245 except NetworkServiceDescriptorError
:
2246 # This could be an NSD not in the nsd-catalog
2249 # Merge any config and initial config primitive values
2250 self
.config_store
.merge_nsd_config(self
.nsd_msg
)
2251 self
._log
.debug("Merged NSD: {}".format(self
.nsd_msg
.as_dict()))
2253 event_descr
= "Fetched NSD with descriptor id %s" % self
.nsd_id
2254 self
.record_event("nsd-fetched", event_descr
)
2256 if self
._nsd
is None:
2257 msg
= "Failed to fetch NSD with nsd-id [%s] for nsr-id %s"
2258 self
._log
.debug(msg
, self
.nsd_id
, self
.id)
2259 raise NetworkServiceRecordError(self
)
2261 self
._log
.debug("Got nsd result %s", self
._nsd
)
2263 # Substitute any input parameters
2264 self
.substitute_input_parameters(self
._nsd
, self
._nsr
_cfg
_msg
)
2267 yield from self
.create(config_xact
)
2269 # Publish the NSR to DTS
2270 yield from self
.publish()
2273 def do_instantiate():
2275 Instantiate network service
2277 self
._log
.debug("Instantiating VLs nsr id [%s] nsd id [%s]",
2278 self
.id, self
.nsd_id
)
2280 # instantiate the VLs
2281 event_descr
= ("Instantiating %s external VLs for NSR id %s" %
2282 (len(self
.nsd_msg
.vld
), self
.id))
2283 self
.record_event("begin-external-vls-instantiation", event_descr
)
2285 self
.set_state(NetworkServiceRecordState
.VL_INIT_PHASE
)
2287 yield from self
.instantiate_vls()
2289 # Publish the NSR to DTS
2290 yield from self
.publish()
2292 event_descr
= ("Finished instantiating %s external VLs for NSR id %s" %
2293 (len(self
.nsd_msg
.vld
), self
.id))
2294 self
.record_event("end-external-vls-instantiation", event_descr
)
2296 self
.set_state(NetworkServiceRecordState
.VNF_INIT_PHASE
)
2298 self
._log
.debug("Instantiating VNFs ...... nsr[%s], nsd[%s]",
2299 self
.id, self
.nsd_id
)
2301 # instantiate the VNFs
2302 event_descr
= ("Instantiating %s VNFS for NSR id %s" %
2303 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2305 self
.record_event("begin-vnf-instantiation", event_descr
)
2307 yield from self
.instantiate_vnfs(self
._vnfrs
.values())
2309 self
._log
.debug(" Finished instantiating %d VNFs for NSR id %s",
2310 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
2312 event_descr
= ("Finished instantiating %s VNFs for NSR id %s" %
2313 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2314 self
.record_event("end-vnf-instantiation", event_descr
)
2316 if len(self
.vnffgrs
) > 0:
2317 #self.set_state(NetworkServiceRecordState.VNFFG_INIT_PHASE)
2318 event_descr
= ("Instantiating %s VNFFGS for NSR id %s" %
2319 (len(self
.nsd_msg
.vnffgd
), self
.id))
2321 self
.record_event("begin-vnffg-instantiation", event_descr
)
2323 yield from self
.instantiate_vnffgs()
2325 event_descr
= ("Finished instantiating %s VNFFGDs for NSR id %s" %
2326 (len(self
.nsd_msg
.vnffgd
), self
.id))
2327 self
.record_event("end-vnffg-instantiation", event_descr
)
2329 if self
.has_scaling_instances():
2330 event_descr
= ("Instantiating %s Scaling Groups for NSR id %s" %
2331 (len(self
._scaling
_groups
), self
.id))
2333 self
.record_event("begin-scaling-group-instantiation", event_descr
)
2334 yield from self
.instantiate_scaling_instances(config_xact
)
2335 self
.record_event("end-scaling-group-instantiation", event_descr
)
2337 # Give the plugin a chance to deploy the network service now that all
2338 # virtual links and vnfs are instantiated
2339 yield from self
.nsm_plugin
.deploy(self
._nsr
_msg
)
2341 self
._log
.debug("Publishing NSR...... nsr[%s], nsd[%s]",
2342 self
.id, self
.nsd_id
)
2344 # Publish the NSR to DTS
2345 yield from self
.publish()
2347 self
._log
.debug("Published NSR...... nsr[%s], nsd[%s]",
2348 self
.id, self
.nsd_id
)
2350 def on_instantiate_done(fut
):
2351 # If the do_instantiate fails, then publish NSR with failed result
2352 if fut
.exception() is not None:
2353 self
._log
.error("NSR instantiation failed for NSR id %s: %s", self
.id, str(fut
.exception()))
2354 self
._loop
.create_task(self
.instantiation_failed(failed_reason
=str(fut
.exception())))
2356 instantiate_task
= self
._loop
.create_task(do_instantiate())
2357 instantiate_task
.add_done_callback(on_instantiate_done
)
2360 def set_config_status(self
, status
, status_details
=None):
2361 if self
.config_status
!= status
:
2362 self
._log
.debug("Updating NSR {} status for {} to {}".
2363 format(self
.name
, self
.config_status
, status
))
2364 self
._config
_status
= status
2365 self
._config
_status
_details
= status_details
2367 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2368 self
.record_event("config-failed", "NS configuration failed",
2369 evt_details
=self
._config
_status
_details
)
2371 yield from self
.publish()
2374 def is_active(self
):
2375 """ This NS is active """
2376 self
.set_state(NetworkServiceRecordState
.RUNNING
)
2380 # Publish the NSR to DTS
2381 self
._log
.debug("Network service %s is active ", self
.id)
2382 self
._is
_active
= True
2384 event_descr
= "NSR in running state for NSR id %s" % self
.id
2385 self
.record_event("ns-running", event_descr
)
2387 yield from self
.publish()
2390 def instantiation_failed(self
, failed_reason
=None):
2391 """ The NS instantiation failed"""
2392 self
._log
.error("Network service id:%s, name:%s instantiation failed",
2394 self
.set_state(NetworkServiceRecordState
.FAILED
)
2396 event_descr
= "Instantiation of NS %s failed" % self
.id
2397 self
.record_event("ns-failed", event_descr
, evt_details
=failed_reason
)
2399 # Publish the NSR to DTS
2400 yield from self
.publish()
2403 def terminate_vnfrs(self
, vnfrs
):
2404 """ Terminate VNFRS in this network service """
2405 self
._log
.debug("Terminating VNFs in network service %s", self
.id)
2407 yield from self
.nsm_plugin
.terminate_vnf(vnfr
)
2410 def terminate(self
):
2411 """ Terminate a NetworkServiceRecord."""
2412 def terminate_vnffgrs():
2413 """ Terminate VNFFGRS in this network service """
2414 self
._log
.debug("Terminating VNFFGRs in network service %s", self
.id)
2415 for vnffgr
in self
.vnffgrs
.values():
2416 yield from vnffgr
.terminate()
2418 def terminate_vlrs():
2419 """ Terminate VLRs in this netork service """
2420 self
._log
.debug("Terminating VLs in network service %s", self
.id)
2421 for vlr
in self
.vlrs
:
2422 yield from self
.nsm_plugin
.terminate_vl(vlr
)
2423 vlr
.state
= VlRecordState
.TERMINATED
2424 if vlr
.id in self
.vlr_uptime_tasks
:
2425 self
.vlr_uptime_tasks
[vlr
.id].cancel()
2427 self
._log
.debug("Terminating network service id %s", self
.id)
2429 # Move the state to TERMINATE
2430 self
.set_state(NetworkServiceRecordState
.TERMINATE
)
2431 event_descr
= "Terminate being processed for NS Id:%s" % self
.id
2432 self
.record_event("terminate", event_descr
)
2434 # Move the state to VNF_TERMINATE_PHASE
2435 self
._log
.debug("Terminating VNFFGs in NS ID: %s", self
.id)
2436 self
.set_state(NetworkServiceRecordState
.VNFFG_TERMINATE_PHASE
)
2437 event_descr
= "Terminating VNFFGS in NS Id:%s" % self
.id
2438 self
.record_event("terminating-vnffgss", event_descr
)
2439 yield from terminate_vnffgrs()
2441 # Move the state to VNF_TERMINATE_PHASE
2442 self
.set_state(NetworkServiceRecordState
.VNF_TERMINATE_PHASE
)
2443 event_descr
= "Terminating VNFS in NS Id:%s" % self
.id
2444 self
.record_event("terminating-vnfs", event_descr
)
2445 yield from self
.terminate_vnfrs(self
.vnfrs
.values())
2447 # Move the state to VL_TERMINATE_PHASE
2448 self
.set_state(NetworkServiceRecordState
.VL_TERMINATE_PHASE
)
2449 event_descr
= "Terminating VLs in NS Id:%s" % self
.id
2450 self
.record_event("terminating-vls", event_descr
)
2451 yield from terminate_vlrs()
2453 yield from self
.nsm_plugin
.terminate_ns(self
)
2455 # Move the state to TERMINATED
2456 self
.set_state(NetworkServiceRecordState
.TERMINATED
)
2457 event_descr
= "Terminated NS Id:%s" % self
.id
2458 self
.record_event("terminated", event_descr
)
2461 """"Enable a NetworkServiceRecord."""
2465 """"Disable a NetworkServiceRecord."""
2468 def map_config_status(self
):
2469 self
._log
.debug("Config status for ns {} is {}".
2470 format(self
.name
, self
._config
_status
))
2471 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURING
:
2472 return 'configuring'
2473 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2477 def vl_phase_completed(self
):
2478 """ Are VLs created in this NS?"""
2479 return self
._vl
_phase
_completed
2481 def vnf_phase_completed(self
):
2482 """ Are VLs created in this NS?"""
2483 return self
._vnf
_phase
_completed
2485 def create_msg(self
):
2486 """ The network serice record as a message """
2487 nsr_dict
= {"ns_instance_config_ref": self
.id}
2488 nsr
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr
.from_dict(nsr_dict
)
2489 #nsr.cloud_account = self.cloud_account_name
2490 nsr
.sdn_account
= self
._sdn
_account
_name
2491 nsr
.name_ref
= self
.name
2492 nsr
.nsd_ref
= self
.nsd_id
2493 nsr
.nsd_name_ref
= self
.nsd_msg
.name
2494 nsr
.operational_events
= self
._op
_status
.msg
2495 nsr
.operational_status
= self
._op
_status
.yang_str()
2496 nsr
.config_status
= self
.map_config_status()
2497 nsr
.config_status_details
= self
._config
_status
_details
2498 nsr
.create_time
= self
._create
_time
2499 nsr
.uptime
= int(time
.time()) - self
._create
_time
2501 for cfg_prim
in self
.nsd_msg
.service_primitive
:
2502 cfg_prim
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_ServicePrimitive
.from_dict(
2504 nsr
.service_primitive
.append(cfg_prim
)
2506 for init_cfg
in self
.nsd_msg
.initial_config_primitive
:
2507 prim
= NsrYang
.NsrInitialConfigPrimitive
.from_dict(
2509 nsr
.initial_config_primitive
.append(prim
)
2511 if self
.vl_phase_completed():
2512 for vlr
in self
.vlrs
:
2513 nsr
.vlr
.append(vlr
.create_nsr_vlr_msg(self
.vnfrs
.values()))
2515 if self
.vnf_phase_completed():
2516 for vnfr_id
in self
.vnfrs
:
2517 nsr
.constituent_vnfr_ref
.append(self
.vnfrs
[vnfr_id
].const_vnfr_msg
)
2518 for vnffgr
in self
.vnffgrs
.values():
2519 nsr
.vnffgr
.append(vnffgr
.fetch_vnffgr())
2520 for scaling_group
in self
._scaling
_groups
.values():
2521 nsr
.scaling_group_record
.append(scaling_group
.create_record_msg())
2525 def all_vnfs_active(self
):
2526 """ Are all VNFS in this NS active? """
2527 for _
, vnfr
in self
.vnfrs
.items():
2528 if vnfr
.active
is not True:
2533 def update_state(self
):
2534 """ Re-evaluate this NS's state """
2535 curr_state
= self
._op
_status
.state
2537 if curr_state
== NetworkServiceRecordState
.TERMINATED
:
2538 self
._log
.debug("NS (%s) in terminated state, not updating state", self
.id)
2541 new_state
= NetworkServiceRecordState
.RUNNING
2542 self
._log
.info("Received update_state for nsr: %s, curr-state: %s",
2543 self
.id, curr_state
)
2545 # Check all the VNFRs are present
2546 for _
, vnfr
in self
.vnfrs
.items():
2547 if vnfr
.state
in [VnfRecordState
.ACTIVE
, VnfRecordState
.TERMINATED
]:
2549 elif vnfr
.state
== VnfRecordState
.FAILED
:
2550 if vnfr
._prev
_state
!= vnfr
.state
:
2551 event_descr
= "Instantiation of VNF %s failed" % vnfr
.id
2552 event_error_details
= vnfr
.state_failed_reason
2553 self
.record_event("vnf-failed", event_descr
, evt_details
=event_error_details
)
2554 vnfr
.set_state(VnfRecordState
.FAILED
)
2556 self
._log
.info("VNF state did not change, curr=%s, prev=%s",
2557 vnfr
.state
, vnfr
._prev
_state
)
2558 new_state
= NetworkServiceRecordState
.FAILED
2561 self
._log
.info("VNF %s in NSR %s is still not active; current state is: %s",
2562 vnfr
.id, self
.id, vnfr
.state
)
2563 new_state
= curr_state
2565 # If new state is RUNNING; check all VLs
2566 if new_state
== NetworkServiceRecordState
.RUNNING
:
2567 for vl
in self
.vlrs
:
2569 if vl
.state
in [VlRecordState
.ACTIVE
, VlRecordState
.TERMINATED
]:
2571 elif vl
.state
== VlRecordState
.FAILED
:
2572 if vl
.prev_state
!= vl
.state
:
2573 event_descr
= "Instantiation of VL %s failed" % vl
.id
2574 event_error_details
= vl
.state_failed_reason
2575 self
.record_event("vl-failed", event_descr
, evt_details
=event_error_details
)
2576 vl
.prev_state
= vl
.state
2578 self
._log
.debug("VL %s already in failed state")
2580 if vl
.state
in [VlRecordState
.INSTANTIATION_PENDING
, VlRecordState
.INIT
]:
2581 new_state
= NetworkServiceRecordState
.VL_INSTANTIATE
2584 if vl
.state
in [VlRecordState
.TERMINATE_PENDING
]:
2585 new_state
= NetworkServiceRecordState
.VL_TERMINATE
2588 # If new state is RUNNING; check VNFFGRs are also active
2589 if new_state
== NetworkServiceRecordState
.RUNNING
:
2590 for _
, vnffgr
in self
.vnffgrs
.items():
2591 self
._log
.info("Checking vnffgr state for nsr %s is: %s",
2592 self
.id, vnffgr
.state
)
2593 if vnffgr
.state
== VnffgRecordState
.ACTIVE
:
2595 elif vnffgr
.state
== VnffgRecordState
.FAILED
:
2596 event_descr
= "Instantiation of VNFFGR %s failed" % vnffgr
.id
2597 self
.record_event("vnffg-failed", event_descr
)
2598 new_state
= NetworkServiceRecordState
.FAILED
2601 self
._log
.info("VNFFGR %s in NSR %s is still not active; current state is: %s",
2602 vnffgr
.id, self
.id, vnffgr
.state
)
2603 new_state
= curr_state
2605 # Update all the scaling group instance operational status to
2606 # reflect the state of all VNFR within that instance
2607 yield from self
._update
_scale
_group
_instances
_status
()
2609 for _
, group
in self
._scaling
_groups
.items():
2610 if group
.state
== scale_group
.ScaleGroupState
.SCALING_OUT
:
2611 new_state
= NetworkServiceRecordState
.SCALING_OUT
2613 elif group
.state
== scale_group
.ScaleGroupState
.SCALING_IN
:
2614 new_state
= NetworkServiceRecordState
.SCALING_IN
2617 if new_state
!= curr_state
:
2618 self
._log
.debug("Changing state of Network service %s from %s to %s",
2619 self
.id, curr_state
, new_state
)
2620 if new_state
== NetworkServiceRecordState
.RUNNING
:
2621 yield from self
.is_active()
2622 elif new_state
== NetworkServiceRecordState
.FAILED
:
2623 # If the NS is already active and we entered scaling_in, scaling_out,
2624 # do not mark the NS as failing if scaling operation failed.
2625 if curr_state
in [NetworkServiceRecordState
.SCALING_OUT
,
2626 NetworkServiceRecordState
.SCALING_IN
] and self
._is
_active
:
2627 new_state
= NetworkServiceRecordState
.RUNNING
2628 self
.set_state(new_state
)
2630 yield from self
.instantiation_failed()
2632 self
.set_state(new_state
)
2634 yield from self
.publish()
2637 class InputParameterSubstitution(object):
2639 This class is responsible for substituting input parameters into an NSD.
2642 def __init__(self
, log
):
2643 """Create an instance of InputParameterSubstitution
2646 log - a logger for this object to use
2651 def __call__(self
, nsd
, nsr_config
):
2652 """Substitutes input parameters from the NSR config into the NSD
2654 This call modifies the provided NSD with the input parameters that are
2655 contained in the NSR config.
2658 nsd - a GI NSD object
2659 nsr_config - a GI NSR config object
2662 if nsd
is None or nsr_config
is None:
2665 # Create a lookup of the xpath elements that this descriptor allows
2667 optional_input_parameters
= set()
2668 for input_parameter
in nsd
.input_parameter_xpath
:
2669 optional_input_parameters
.add(input_parameter
.xpath
)
2671 # Apply the input parameters to the descriptor
2672 if nsr_config
.input_parameter
:
2673 for param
in nsr_config
.input_parameter
:
2674 if param
.xpath
not in optional_input_parameters
:
2675 msg
= "tried to set an invalid input parameter ({})"
2676 self
.log
.error(msg
.format(param
.xpath
))
2680 "input-parameter:{} = {}".format(
2687 xpath
.setxattr(nsd
, param
.xpath
, param
.value
)
2689 except Exception as e
:
2690 self
.log
.exception(e
)
2693 class NetworkServiceDescriptor(object):
2695 Network service descriptor class
2698 def __init__(self
, dts
, log
, loop
, nsd
, nsm
):
2710 """ Returns nsd id """
2715 """ Returns name of nsd """
2716 return self
._nsd
.name
2719 def ref_count(self
):
2720 """ Returns reference count"""
2721 return self
._ref
_count
2724 """ Returns whether nsd is in use or not """
2725 return True if self
.ref_count
> 0 else False
2728 """ Take a reference on this object """
2729 self
._ref
_count
+= 1
2732 """ Release reference on this object """
2733 if self
.ref_count
< 1:
2734 msg
= ("Unref on a NSD object - nsd id %s, ref_count = %s" %
2735 (self
.id, self
.ref_count
))
2736 self
._log
.critical(msg
)
2737 raise NetworkServiceDescriptorError(msg
)
2738 self
._ref
_count
-= 1
2742 """ Return the message associated with this NetworkServiceDescriptor"""
2746 def path_for_id(nsd_id
):
2747 """ Return path for the passed nsd_id"""
2748 return "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}'".format(nsd_id
)
2751 """ Return the message associated with this NetworkServiceDescriptor"""
2752 return NetworkServiceDescriptor
.path_for_id(self
.id)
2754 def update(self
, nsd
):
2755 """ Update the NSD descriptor """
2759 class NsdDtsHandler(object):
2760 """ The network service descriptor DTS handler """
2761 XPATH
= "C,/nsd:nsd-catalog/nsd:nsd"
2763 def __init__(self
, dts
, log
, loop
, nsm
):
2773 """ Return registration handle """
2778 """ Register for Nsd create/update/delete/read requests from dts """
2780 def on_apply(dts
, acg
, xact
, action
, scratch
):
2781 """Apply the configuration"""
2782 is_recovery
= xact
.xact
is None and action
== rwdts
.AppconfAction
.INSTALL
2783 self
._log
.debug("Got nsd apply cfg (xact:%s) (action:%s)",
2785 # Create/Update an NSD record
2786 for cfg
in self
._regh
.get_xact_elements(xact
):
2787 # Only interested in those NSD cfgs whose ID was received in prepare callback
2788 if cfg
.id in scratch
.get('nsds', []) or is_recovery
:
2789 self
._nsm
.update_nsd(cfg
)
2791 scratch
.pop('nsds', None)
2793 return RwTypes
.RwStatus
.SUCCESS
2796 def delete_nsd_libs(nsd_id
):
2797 """ Remove any files uploaded with NSD and stored under $RIFT_ARTIFACTS/libs/<id> """
2799 rift_artifacts_dir
= os
.environ
['RIFT_ARTIFACTS']
2800 nsd_dir
= os
.path
.join(rift_artifacts_dir
, 'launchpad/libs', nsd_id
)
2802 if os
.path
.exists (nsd_dir
):
2803 shutil
.rmtree(nsd_dir
, ignore_errors
=True)
2804 except Exception as e
:
2805 self
._log
.error("Exception in cleaning up NSD libs {}: {}".
2807 self
._log
.excpetion(e
)
2810 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2811 """ Prepare callback from DTS for NSD config """
2813 self
._log
.info("Got nsd prepare - config received nsd id %s, msg %s",
2816 fref
= ProtobufC
.FieldReference
.alloc()
2817 fref
.goto_whole_message(msg
.to_pbcm())
2819 if fref
.is_field_deleted():
2820 # Delete an NSD record
2821 self
._log
.debug("Deleting NSD with id %s", msg
.id)
2822 if self
._nsm
.nsd_in_use(msg
.id):
2823 self
._log
.debug("Cannot delete NSD in use - %s", msg
.id)
2824 err
= "Cannot delete an NSD in use - %s" % msg
.id
2825 raise NetworkServiceDescriptorRefCountExists(err
)
2827 yield from delete_nsd_libs(msg
.id)
2828 self
._nsm
.delete_nsd(msg
.id)
2830 # Add this NSD to scratch to create/update in apply callback
2831 nsds
= scratch
.setdefault('nsds', [])
2833 # acg._scratch['nsds'].append(msg.id)
2835 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2838 "Registering for NSD config using xpath: %s",
2839 NsdDtsHandler
.XPATH
,
2842 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2843 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2844 # Need a list in scratch to store NSDs to create/update later
2845 # acg._scratch['nsds'] = list()
2846 self
._regh
= acg
.register(
2847 xpath
=NsdDtsHandler
.XPATH
,
2848 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
2849 on_prepare
=on_prepare
)
2852 class VnfdDtsHandler(object):
2853 """ DTS handler for VNFD config changes """
2854 XPATH
= "C,/vnfd:vnfd-catalog/vnfd:vnfd"
2856 def __init__(self
, dts
, log
, loop
, nsm
):
2865 """ DTS registration handle """
2870 """ Register for VNFD configuration"""
2873 def on_apply(dts
, acg
, xact
, action
, scratch
):
2874 """Apply the configuration"""
2875 self
._log
.debug("Got NSM VNFD apply (xact: %s) (action: %s)(scr: %s)",
2876 xact
, action
, scratch
)
2878 # Create/Update a VNFD record
2879 for cfg
in self
._regh
.get_xact_elements(xact
):
2880 # Only interested in those VNFD cfgs whose ID was received in prepare callback
2881 if cfg
.id in scratch
.get('vnfds', []):
2882 self
._nsm
.update_vnfd(cfg
)
2884 for cfg
in self
._regh
.elements
:
2885 if cfg
.id in scratch
.get('deleted_vnfds', []):
2886 yield from self
._nsm
.delete_vnfd(cfg
.id)
2888 scratch
.pop('vnfds', None)
2889 scratch
.pop('deleted_vnfds', None)
2892 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2893 """ on prepare callback """
2894 self
._log
.debug("Got on prepare for VNFD (path: %s) (action: %s) (msg: %s)",
2895 ks_path
.to_xpath(RwNsmYang
.get_schema()), xact_info
.query_action
, msg
)
2897 fref
= ProtobufC
.FieldReference
.alloc()
2898 fref
.goto_whole_message(msg
.to_pbcm())
2900 # Handle deletes in prepare_callback, but adds/updates in apply_callback
2901 if fref
.is_field_deleted():
2902 self
._log
.debug("Adding msg to deleted field")
2903 deleted_vnfds
= scratch
.setdefault('deleted_vnfds', [])
2904 deleted_vnfds
.append(msg
.id)
2906 # Add this VNFD to scratch to create/update in apply callback
2907 vnfds
= scratch
.setdefault('vnfds', [])
2908 vnfds
.append(msg
.id)
2910 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2913 "Registering for VNFD config using xpath: %s",
2914 VnfdDtsHandler
.XPATH
,
2916 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2917 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2918 # Need a list in scratch to store VNFDs to create/update later
2919 # acg._scratch['vnfds'] = list()
2920 # acg._scratch['deleted_vnfds'] = list()
2921 self
._regh
= acg
.register(
2922 xpath
=VnfdDtsHandler
.XPATH
,
2923 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY
,
2924 on_prepare
=on_prepare
)
2926 class NsrRpcDtsHandler(object):
2927 """ The network service instantiation RPC DTS handler """
2928 EXEC_NSR_CONF_XPATH
= "I,/nsr:start-network-service"
2929 EXEC_NSR_CONF_O_XPATH
= "O,/nsr:start-network-service"
2930 NETCONF_IP_ADDRESS
= "127.0.0.1"
2932 RESTCONF_PORT
= 8888
2933 NETCONF_USER
= "admin"
2934 NETCONF_PW
= "admin"
2935 REST_BASE_V2_URL
= 'https://{}:{}/v2/api/'.format("127.0.0.1",8888)
2937 def __init__(self
, dts
, log
, loop
, nsm
):
2944 self
._ns
_regh
= None
2946 self
._manager
= None
2947 self
._nsr
_config
_url
= NsrRpcDtsHandler
.REST_BASE_V2_URL
+ 'config/ns-instance-config'
2949 self
._model
= RwYang
.Model
.create_libncx()
2950 self
._model
.load_schema_ypbc(RwNsrYang
.get_schema())
2954 """ Return the NS manager instance """
2958 def wrap_netconf_config_xml(xml
):
2959 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
2963 def _connect(self
, timeout_secs
=240):
2965 start_time
= time
.time()
2966 while (time
.time() - start_time
) < timeout_secs
:
2969 self
._log
.debug("Attemping NsmTasklet netconf connection.")
2971 manager
= yield from ncclient
.asyncio_manager
.asyncio_connect(
2973 host
=NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
,
2974 port
=NsrRpcDtsHandler
.NETCONF_PORT
,
2975 username
=NsrRpcDtsHandler
.NETCONF_USER
,
2976 password
=NsrRpcDtsHandler
.NETCONF_PW
,
2978 look_for_keys
=False,
2979 hostkey_verify
=False,
2984 except ncclient
.transport
.errors
.SSHError
as e
:
2985 self
._log
.warning("Netconf connection to launchpad %s failed: %s",
2986 NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
, str(e
))
2988 yield from asyncio
.sleep(5, loop
=self
._loop
)
2990 raise NsrInstantiationFailed("Failed to connect to Launchpad within %s seconds" %
2993 def _apply_ns_instance_config(self
,payload_dict
):
2994 #self._log.debug("At apply NS instance config with payload %s",payload_dict)
2995 req_hdr
= {'accept':'application/vnd.yang.data+json','content-type':'application/vnd.yang.data+json'}
2996 response
=requests
.post(self
._nsr
_config
_url
, headers
=req_hdr
, auth
=('admin', 'admin'),data
=payload_dict
,verify
=False)
3001 """ Register for NS monitoring read from dts """
3003 def on_ns_config_prepare(xact_info
, action
, ks_path
, msg
):
3004 """ prepare callback from dts start-network-service"""
3005 assert action
== rwdts
.QueryAction
.RPC
3007 rpc_op
= NsrYang
.YangOutput_Nsr_StartNetworkService
.from_dict({
3008 "nsr_id":str(uuid
.uuid4())
3011 if not ('name' in rpc_ip
and 'nsd_ref' in rpc_ip
and ('cloud_account' in rpc_ip
or 'om_datacenter' in rpc_ip
)):
3012 self
._log
.error("Mandatory parameters name or nsd_ref or cloud account not found in start-network-service {}".format(rpc_ip
))
3015 self
._log
.debug("start-network-service RPC input: {}".format(rpc_ip
))
3018 # Add used value to the pool
3019 self
._log
.debug("RPC output: {}".format(rpc_op
))
3021 nsd_copy
= self
.nsm
.get_nsd(rpc_ip
.nsd_ref
)
3023 #if not self._manager:
3024 # self._manager = yield from self._connect()
3026 self
._log
.debug("Configuring ns-instance-config with name %s nsd-ref: %s",
3027 rpc_ip
.name
, rpc_ip
.nsd_ref
)
3029 ns_instance_config_dict
= {"id":rpc_op
.nsr_id
, "admin_status":"ENABLED"}
3030 ns_instance_config_copy_dict
= {k
:v
for k
, v
in rpc_ip
.as_dict().items()
3031 if k
in RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr().fields
}
3032 ns_instance_config_dict
.update(ns_instance_config_copy_dict
)
3034 ns_instance_config
= RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.from_dict(ns_instance_config_dict
)
3035 ns_instance_config
.nsd
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_Nsd()
3036 ns_instance_config
.nsd
.from_dict(nsd_copy
.msg
.as_dict())
3038 payload_dict
= ns_instance_config
.to_json(self
._model
)
3039 #xml = ns_instance_config.to_xml_v2(self._model)
3040 #netconf_xml = self.wrap_netconf_config_xml(xml)
3042 #self._log.debug("Sending configure ns-instance-config xml to %s: %s",
3043 # netconf_xml, NsrRpcDtsHandler.NETCONF_IP_ADDRESS)
3044 self
._log
.debug("Sending configure ns-instance-config json to %s: %s",
3045 self
._nsr
_config
_url
,ns_instance_config
)
3047 #response = yield from self._manager.edit_config(
3049 # config=netconf_xml,
3051 response
= yield from self
._loop
.run_in_executor(
3053 self
._apply
_ns
_instance
_config
,
3056 response
.raise_for_status()
3057 self
._log
.debug("Received edit config response: %s", response
.json())
3059 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
,
3060 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
,
3062 except Exception as e
:
3063 self
._log
.error("Exception processing the "
3064 "start-network-service: {}".format(e
))
3065 self
._log
.exception(e
)
3066 xact_info
.respond_xpath(rwdts
.XactRspCode
.NACK
,
3067 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
)
3070 hdl_ns
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_ns_config_prepare
,)
3072 with self
._dts
.group_create() as group
:
3073 self
._ns
_regh
= group
.register(xpath
=NsrRpcDtsHandler
.EXEC_NSR_CONF_XPATH
,
3075 flags
=rwdts
.Flag
.PUBLISHER
,
3079 class NsrDtsHandler(object):
3080 """ The network service DTS handler """
3081 NSR_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr"
3082 SCALE_INSTANCE_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr/nsr:scaling-group/nsr:instance"
3083 KEY_PAIR_XPATH
= "C,/nsr:key-pair"
3085 def __init__(self
, dts
, log
, loop
, nsm
):
3091 self
._nsr
_regh
= None
3092 self
._scale
_regh
= None
3093 self
._key
_pair
_regh
= None
3097 """ Return the NS manager instance """
3102 """ Register for Nsr create/update/delete/read requests from dts """
3104 def nsr_id_from_keyspec(ks
):
3105 nsr_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.schema().keyspec_to_entry(ks
)
3106 nsr_id
= nsr_path_entry
.key00
.id
3109 def group_name_from_keyspec(ks
):
3110 group_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
.schema().keyspec_to_entry(ks
)
3111 group_name
= group_path_entry
.key00
.scaling_group_name_ref
3114 def is_instance_in_reg_elements(nsr_id
, group_name
, instance_id
):
3115 """ Return boolean indicating if scaling group instance was already commited previously.
3117 By looking at the existing elements in this registration handle (elements not part
3118 of this current xact), we can tell if the instance was configured previously without
3119 keeping any application state.
3121 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3122 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3123 elem_group_name
= group_name_from_keyspec(keyspec
)
3125 if elem_nsr_id
!= nsr_id
or group_name
!= elem_group_name
:
3128 if instance_cfg
.id == instance_id
:
3133 def get_scale_group_instance_delta(nsr_id
, group_name
, xact
):
3134 delta
= {"added": [], "deleted": []}
3135 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3136 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3137 if elem_nsr_id
!= nsr_id
:
3140 elem_group_name
= group_name_from_keyspec(keyspec
)
3141 if elem_group_name
!= group_name
:
3144 delta
["added"].append(instance_cfg
.id)
3146 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(include_keyspec
=True):
3147 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3148 if elem_nsr_id
!= nsr_id
:
3151 elem_group_name
= group_name_from_keyspec(keyspec
)
3152 if elem_group_name
!= group_name
:
3155 if instance_cfg
.id in delta
["added"]:
3156 delta
["added"].remove(instance_cfg
.id)
3158 delta
["deleted"].append(instance_cfg
.id)
3163 def update_nsr_nsd(nsr_id
, xact
, scratch
):
3166 def get_nsr_vl_delta(nsr_id
, xact
, scratch
):
3167 delta
= {"added": [], "deleted": []}
3168 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3169 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3170 if elem_nsr_id
!= nsr_id
:
3173 if 'vld' in instance_cfg
.nsd
:
3174 for vld
in instance_cfg
.nsd
.vld
:
3175 delta
["added"].append(vld
)
3177 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3178 self
._log
.debug("NSR update: %s", instance_cfg
)
3179 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3180 if elem_nsr_id
!= nsr_id
:
3183 if 'vld' in instance_cfg
.nsd
:
3184 for vld
in instance_cfg
.nsd
.vld
:
3185 if vld
in delta
["added"]:
3186 delta
["added"].remove(vld
)
3188 delta
["deleted"].append(vld
)
3192 vl_delta
= yield from get_nsr_vl_delta(nsr_id
, xact
, scratch
)
3193 self
._log
.debug("Got NSR:%s VL instance delta: %s", nsr_id
, vl_delta
)
3195 for vld
in vl_delta
["added"]:
3196 yield from self
._nsm
.nsr_instantiate_vl(nsr_id
, vld
)
3198 for vld
in vl_delta
["deleted"]:
3199 yield from self
._nsm
.nsr_terminate_vl(nsr_id
, vld
)
3201 def get_add_delete_update_cfgs(dts_member_reg
, xact
, key_name
, scratch
):
3202 # Unfortunately, it is currently difficult to figure out what has exactly
3203 # changed in this xact without Pbdelta support (RIFT-4916)
3204 # As a workaround, we can fetch the pre and post xact elements and
3205 # perform a comparison to figure out adds/deletes/updates
3206 xact_cfgs
= list(dts_member_reg
.get_xact_elements(xact
))
3207 curr_cfgs
= list(dts_member_reg
.elements
)
3209 xact_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in xact_cfgs
}
3210 curr_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in curr_cfgs
}
3213 added_keys
= set(xact_key_map
) - set(curr_key_map
)
3214 added_cfgs
= [xact_key_map
[key
] for key
in added_keys
]
3217 deleted_keys
= set(curr_key_map
) - set(xact_key_map
)
3218 deleted_cfgs
= [curr_key_map
[key
] for key
in deleted_keys
]
3221 updated_keys
= set(curr_key_map
) & set(xact_key_map
)
3222 updated_cfgs
= [xact_key_map
[key
] for key
in updated_keys
3223 if xact_key_map
[key
] != curr_key_map
[key
]]
3225 return added_cfgs
, deleted_cfgs
, updated_cfgs
3227 def get_nsr_key_pairs(dts_member_reg
, xact
):
3229 for instance_cfg
, keyspec
in dts_member_reg
.get_xact_elements(xact
, include_keyspec
=True):
3230 self
._log
.debug("Key pair received is {} KS: {}".format(instance_cfg
, keyspec
))
3231 xpath
= keyspec
.to_xpath(RwNsrYang
.get_schema())
3232 key_pairs
[instance_cfg
.name
] = instance_cfg
3235 def on_apply(dts
, acg
, xact
, action
, scratch
):
3236 """Apply the configuration"""
3237 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3238 xact
, action
, scratch
)
3240 def handle_create_nsr(msg
, key_pairs
=None, restart_mode
=False):
3241 # Handle create nsr requests """
3242 # Do some validations
3243 if not msg
.has_field("nsd"):
3244 err
= "NSD not provided"
3245 self
._log
.error(err
)
3246 raise NetworkServiceRecordError(err
)
3248 self
._log
.debug("Creating NetworkServiceRecord %s from nsr config %s",
3249 msg
.id, msg
.as_dict())
3250 nsr
= self
.nsm
.create_nsr(msg
, key_pairs
=key_pairs
, restart_mode
=restart_mode
)
3253 def handle_delete_nsr(msg
):
3255 def delete_instantiation(ns_id
):
3256 """ Delete instantiation """
3257 with self
._dts
.transaction() as xact
:
3258 yield from self
._nsm
.terminate_ns(ns_id
, xact
)
3260 # Handle delete NSR requests
3261 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3262 # Terminate the NSR instance
3263 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3265 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3266 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3267 nsr
.record_event("terminate-rcvd", event_descr
)
3269 self
._loop
.create_task(delete_instantiation(msg
.id))
3272 def begin_instantiation(nsr
):
3273 # Begin instantiation
3274 self
._log
.info("Beginning NS instantiation: %s", nsr
.id)
3275 yield from self
._nsm
.instantiate_ns(nsr
.id, xact
)
3277 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3278 xact
, action
, scratch
)
3280 if action
== rwdts
.AppconfAction
.INSTALL
and xact
.id is None:
3282 for element
in self
._key
_pair
_regh
.elements
:
3283 key_pairs
.append(element
)
3284 for element
in self
._nsr
_regh
.elements
:
3285 nsr
= handle_create_nsr(element
, key_pairs
, restart_mode
=True)
3286 self
._loop
.create_task(begin_instantiation(nsr
))
3289 (added_msgs
, deleted_msgs
, updated_msgs
) = get_add_delete_update_cfgs(self
._nsr
_regh
,
3293 self
._log
.debug("Added: %s, Deleted: %s, Updated: %s", added_msgs
,
3294 deleted_msgs
, updated_msgs
)
3296 for msg
in added_msgs
:
3297 if msg
.id not in self
._nsm
.nsrs
:
3298 self
._log
.info("Create NSR received in on_apply to instantiate NS:%s", msg
.id)
3299 key_pairs
= get_nsr_key_pairs(self
._key
_pair
_regh
, xact
)
3300 nsr
= handle_create_nsr(msg
,key_pairs
)
3301 self
._loop
.create_task(begin_instantiation(nsr
))
3303 for msg
in deleted_msgs
:
3304 self
._log
.info("Delete NSR received in on_apply to terminate NS:%s", msg
.id)
3306 handle_delete_nsr(msg
)
3308 self
._log
.exception("Failed to terminate NS:%s", msg
.id)
3310 for msg
in updated_msgs
:
3311 self
._log
.info("Update NSR received in on_apply: %s", msg
)
3313 self
._nsm
.nsr_update_cfg(msg
.id, msg
)
3316 self
._loop
.create_task(update_nsr_nsd(msg
.id, xact
, scratch
))
3318 for group
in msg
.scaling_group
:
3319 instance_delta
= get_scale_group_instance_delta(msg
.id, group
.scaling_group_name_ref
, xact
)
3320 self
._log
.debug("Got NSR:%s scale group instance delta: %s", msg
.id, instance_delta
)
3322 for instance_id
in instance_delta
["added"]:
3323 self
._nsm
.scale_nsr_out(msg
.id, group
.scaling_group_name_ref
, instance_id
, xact
)
3325 for instance_id
in instance_delta
["deleted"]:
3326 self
._nsm
.scale_nsr_in(msg
.id, group
.scaling_group_name_ref
, instance_id
)
3329 return RwTypes
.RwStatus
.SUCCESS
3332 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
3333 """ Prepare calllback from DTS for NSR """
3335 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3336 action
= xact_info
.query_action
3338 "Got Nsr prepare callback (xact: %s) (action: %s) (info: %s), %s:%s)",
3339 xact
, action
, xact_info
, xpath
, msg
3343 def delete_instantiation(ns_id
):
3344 """ Delete instantiation """
3345 yield from self
._nsm
.terminate_ns(ns_id
, None)
3347 def handle_delete_nsr():
3348 """ Handle delete NSR requests """
3349 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3350 # Terminate the NSR instance
3351 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3353 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3354 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3355 nsr
.record_event("terminate-rcvd", event_descr
)
3357 self
._loop
.create_task(delete_instantiation(msg
.id))
3359 fref
= ProtobufC
.FieldReference
.alloc()
3360 fref
.goto_whole_message(msg
.to_pbcm())
3362 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
, rwdts
.QueryAction
.DELETE
]:
3363 # if this is an NSR create
3364 if action
!= rwdts
.QueryAction
.DELETE
and msg
.id not in self
._nsm
.nsrs
:
3365 # Ensure the Cloud account/datacenter has been specified
3366 if not msg
.has_field("cloud_account") and not msg
.has_field("om_datacenter"):
3367 raise NsrInstantiationFailed("Cloud account or datacenter not specified in NSR")
3369 # Check if nsd is specified
3370 if not msg
.has_field("nsd"):
3371 raise NsrInstantiationFailed("NSD not specified in NSR")
3374 nsr
= self
._nsm
.nsrs
[msg
.id]
3376 if msg
.has_field("nsd"):
3377 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3378 raise NsrVlUpdateError("Unable to update VL when NSR not in running state")
3379 if 'vld' not in msg
.nsd
or len(msg
.nsd
.vld
) == 0:
3380 raise NsrVlUpdateError("NS config NSD should have atleast 1 VLD defined")
3382 if msg
.has_field("scaling_group"):
3383 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3384 raise ScalingOperationError("Unable to perform scaling action when NS is not in running state")
3386 if len(msg
.scaling_group
) > 1:
3387 raise ScalingOperationError("Only a single scaling group can be configured at a time")
3389 for group_msg
in msg
.scaling_group
:
3390 num_new_group_instances
= len(group_msg
.instance
)
3391 if num_new_group_instances
> 1:
3392 raise ScalingOperationError("Only a single scaling instance can be modified at a time")
3394 elif num_new_group_instances
== 1:
3395 scale_group
= nsr
.scaling_groups
[group_msg
.scaling_group_name_ref
]
3396 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
]:
3397 if len(scale_group
.instances
) == scale_group
.max_instance_count
:
3398 raise ScalingOperationError("Max instances for %s reached" % scale_group
)
3400 acg
.handle
.prepare_complete_ok(xact_info
.handle
)
3403 self
._log
.debug("Registering for NSR config using xpath: %s",
3404 NsrDtsHandler
.NSR_XPATH
)
3406 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
3407 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
3408 self
._nsr
_regh
= acg
.register(xpath
=NsrDtsHandler
.NSR_XPATH
,
3409 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3410 on_prepare
=on_prepare
)
3412 self
._scale
_regh
= acg
.register(
3413 xpath
=NsrDtsHandler
.SCALE_INSTANCE_XPATH
,
3414 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY| rwdts
.Flag
.CACHE
,
3417 self
._key
_pair
_regh
= acg
.register(
3418 xpath
=NsrDtsHandler
.KEY_PAIR_XPATH
,
3419 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3423 class NsrOpDataDtsHandler(object):
3424 """ The network service op data DTS handler """
3425 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
3427 def __init__(self
, dts
, log
, loop
, nsm
):
3436 """ Return the registration handle"""
3441 """ Return the NS manager instance """
3446 """ Register for Nsr op data publisher registration"""
3447 self
._log
.debug("Registering Nsr op data path %s as publisher",
3448 NsrOpDataDtsHandler
.XPATH
)
3450 hdl
= rift
.tasklets
.DTS
.RegistrationHandler()
3451 handlers
= rift
.tasklets
.Group
.Handler()
3452 with self
._dts
.group_create(handler
=handlers
) as group
:
3453 self
._regh
= group
.register(xpath
=NsrOpDataDtsHandler
.XPATH
,
3455 flags
=rwdts
.Flag
.PUBLISHER | rwdts
.Flag
.NO_PREP_READ | rwdts
.Flag
.DATASTORE
)
3458 def create(self
, path
, msg
):
3460 Create an NS record in DTS with the path and message
3462 self
._log
.debug("Creating NSR %s:%s", path
, msg
)
3463 self
.regh
.create_element(path
, msg
)
3464 self
._log
.debug("Created NSR, %s:%s", path
, msg
)
3467 def update(self
, path
, msg
, flags
=rwdts
.XactFlag
.REPLACE
):
3469 Update an NS record in DTS with the path and message
3471 self
._log
.debug("Updating NSR, %s:%s regh = %s", path
, msg
, self
.regh
)
3472 self
.regh
.update_element(path
, msg
, flags
)
3473 self
._log
.debug("Updated NSR, %s:%s", path
, msg
)
3476 def delete(self
, path
):
3478 Update an NS record in DTS with the path and message
3480 self
._log
.debug("Deleting NSR path:%s", path
)
3481 self
.regh
.delete_element(path
)
3482 self
._log
.debug("Deleted NSR path:%s", path
)
3485 class VnfrDtsHandler(object):
3486 """ The virtual network service DTS handler """
3487 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
3489 def __init__(self
, dts
, log
, loop
, nsm
):
3499 """ Return registration handle """
3504 """ Return the NS manager instance """
3509 """ Register for vnfr create/update/delete/ advises from dts """
3511 def on_commit(xact_info
):
3512 """ The transaction has been committed """
3513 self
._log
.debug("Got vnfr commit (xact_info: %s)", xact_info
)
3514 return rwdts
.MemberRspCode
.ACTION_OK
3517 def on_prepare(xact_info
, action
, ks_path
, msg
):
3518 """ prepare callback from dts """
3519 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3521 "Got vnfr on_prepare cb (xact_info: %s, action: %s): %s:%s",
3522 xact_info
, action
, ks_path
, msg
3525 schema
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr
.schema()
3526 path_entry
= schema
.keyspec_to_entry(ks_path
)
3527 if path_entry
.key00
.id not in self
._nsm
._vnfrs
:
3528 self
._log
.error("%s request for non existent record path %s",
3530 xact_info
.respond_xpath(rwdts
.XactRspCode
.NA
, xpath
)
3534 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3535 if action
== rwdts
.QueryAction
.CREATE
or action
== rwdts
.QueryAction
.UPDATE
:
3536 yield from self
._nsm
.update_vnfr(msg
)
3537 elif action
== rwdts
.QueryAction
.DELETE
:
3538 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3539 self
._nsm
.delete_vnfr(path_entry
.key00
.id)
3541 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
, xpath
)
3543 self
._log
.debug("Registering for VNFR using xpath: %s",
3544 VnfrDtsHandler
.XPATH
,)
3546 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_commit
=on_commit
,
3547 on_prepare
=on_prepare
,)
3548 with self
._dts
.group_create() as group
:
3549 self
._regh
= group
.register(xpath
=VnfrDtsHandler
.XPATH
,
3551 flags
=(rwdts
.Flag
.SUBSCRIBER
),)
3554 class NsdRefCountDtsHandler(object):
3555 """ The NSD Ref Count DTS handler """
3556 XPATH
= "D,/nsr:ns-instance-opdata/rw-nsr:nsd-ref-count"
3558 def __init__(self
, dts
, log
, loop
, nsm
):
3568 """ Return registration handle """
3573 """ Return the NS manager instance """
3578 """ Register for NSD ref count read from dts """
3581 def on_prepare(xact_info
, action
, ks_path
, msg
):
3582 """ prepare callback from dts """
3583 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3585 if action
== rwdts
.QueryAction
.READ
:
3586 schema
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount
.schema()
3587 path_entry
= schema
.keyspec_to_entry(ks_path
)
3588 nsd_list
= yield from self
._nsm
.get_nsd_refcount(path_entry
.key00
.nsd_id_ref
)
3589 for xpath
, msg
in nsd_list
:
3590 xact_info
.respond_xpath(rsp_code
=rwdts
.XactRspCode
.MORE
,
3593 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
3595 raise NetworkServiceRecordError("Not supported operation %s" % action
)
3597 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_prepare
,)
3598 with self
._dts
.group_create() as group
:
3599 self
._regh
= group
.register(xpath
=NsdRefCountDtsHandler
.XPATH
,
3601 flags
=rwdts
.Flag
.PUBLISHER
,)
3604 class NsManager(object):
3605 """ The Network Service Manager class"""
3606 def __init__(self
, dts
, log
, loop
,
3607 nsr_handler
, vnfr_handler
, vlr_handler
, ro_plugin_selector
,
3608 vnffgmgr
, vnfd_pub_handler
, cloud_account_handler
):
3612 self
._nsr
_handler
= nsr_handler
3613 self
._vnfr
_pub
_handler
= vnfr_handler
3614 self
._vlr
_pub
_handler
= vlr_handler
3615 self
._vnffgmgr
= vnffgmgr
3616 self
._vnfd
_pub
_handler
= vnfd_pub_handler
3617 self
._cloud
_account
_handler
= cloud_account_handler
3619 self
._ro
_plugin
_selector
= ro_plugin_selector
3620 self
._ncclient
= rift
.mano
.ncclient
.NcClient(
3632 self
.cfgmgr_obj
= conman
.ROConfigManager(log
, loop
, dts
, self
)
3634 # TODO: All these handlers should move to tasklet level.
3635 # Passing self is often an indication of bad design
3636 self
._nsd
_dts
_handler
= NsdDtsHandler(dts
, log
, loop
, self
)
3637 self
._vnfd
_dts
_handler
= VnfdDtsHandler(dts
, log
, loop
, self
)
3638 self
._dts
_handlers
= [self
._nsd
_dts
_handler
,
3639 VnfrDtsHandler(dts
, log
, loop
, self
),
3640 NsdRefCountDtsHandler(dts
, log
, loop
, self
),
3641 NsrDtsHandler(dts
, log
, loop
, self
),
3642 ScalingRpcHandler(log
, dts
, loop
, self
.scale_rpc_callback
),
3643 NsrRpcDtsHandler(dts
,log
,loop
,self
),
3644 self
._vnfd
_dts
_handler
,
3665 def nsr_handler(self
):
3666 """" NSR handler """
3667 return self
._nsr
_handler
3671 """" So Obj handler """
3676 """ NSRs in this NSM"""
3681 """ NSDs in this NSM"""
3686 """ VNFDs in this NSM"""
3691 """ VNFRs in this NSM"""
3695 def nsr_pub_handler(self
):
3696 """ NSR publication handler """
3697 return self
._nsr
_handler
3700 def vnfr_pub_handler(self
):
3701 """ VNFR publication handler """
3702 return self
._vnfr
_pub
_handler
3705 def vlr_pub_handler(self
):
3706 """ VLR publication handler """
3707 return self
._vlr
_pub
_handler
3710 def vnfd_pub_handler(self
):
3711 return self
._vnfd
_pub
_handler
3715 """ Register all static DTS handlers """
3716 for dts_handle
in self
._dts
_handlers
:
3717 yield from dts_handle
.register()
3720 def get_ns_by_nsr_id(self
, nsr_id
):
3721 """ get NSR by nsr id """
3722 if nsr_id
not in self
._nsrs
:
3723 raise NetworkServiceRecordError("NSR id %s not found" % nsr_id
)
3725 return self
._nsrs
[nsr_id
]
3727 def scale_nsr_out(self
, nsr_id
, scale_group_name
, instance_id
, config_xact
):
3728 self
.log
.debug("Scale out NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3733 nsr
= self
._nsrs
[nsr_id
]
3734 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3735 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3737 self
._loop
.create_task(nsr
.create_scale_group_instance(scale_group_name
, instance_id
, config_xact
))
3739 def scale_nsr_in(self
, nsr_id
, scale_group_name
, instance_id
):
3740 self
.log
.debug("Scale in NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3745 nsr
= self
._nsrs
[nsr_id
]
3746 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3747 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3749 self
._loop
.create_task(nsr
.delete_scale_group_instance(scale_group_name
, instance_id
))
3751 def scale_rpc_callback(self
, xact
, msg
, action
):
3752 """Callback handler for RPC calls
3754 xact : Transaction Handler
3756 action : Scaling Action
3758 ScalingGroupInstance
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup_Instance
3759 ScalingGroup
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
3761 xpath
= ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').format(
3763 instance
= ScalingGroupInstance
.from_dict({"id": msg
.instance_id
})
3766 def get_nsr_scaling_group():
3767 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3769 for result
in results
:
3770 res
= yield from result
3771 nsr_config
= res
.result
3773 for scaling_group
in nsr_config
.scaling_group
:
3774 if scaling_group
.scaling_group_name_ref
== msg
.scaling_group_name_ref
:
3777 scaling_group
= nsr_config
.scaling_group
.add()
3778 scaling_group
.scaling_group_name_ref
= msg
.scaling_group_name_ref
3780 return (nsr_config
, scaling_group
)
3783 def update_config(nsr_config
):
3784 xml
= self
._ncclient
.convert_to_xml(RwNsrYang
, nsr_config
)
3785 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
3786 yield from self
._ncclient
.connect()
3787 yield from self
._ncclient
.manager
.edit_config(target
="running", config
=xml
, default_operation
="replace")
3791 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3792 scaling_group
.instance
.append(instance
)
3793 yield from update_config(nsr_config
)
3797 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3798 scaling_group
.instance
.remove(instance
)
3799 yield from update_config(nsr_config
)
3801 if action
== ScalingRpcHandler
.ACTION
.SCALE_OUT
:
3802 self
._loop
.create_task(scale_out())
3804 self
._loop
.create_task(scale_in())
3806 # Opdata based calls, disabled for now!
3807 # if action == ScalingRpcHandler.ACTION.SCALE_OUT:
3808 # self.scale_nsr_out(
3810 # msg.scaling_group_name_ref,
3814 # self.scale_nsr_in(
3816 # msg.scaling_group_name_ref,
3819 def nsr_update_cfg(self
, nsr_id
, msg
):
3820 nsr
= self
._nsrs
[nsr_id
]
3821 nsr
.nsr_cfg_msg
= msg
3823 def nsr_instantiate_vl(self
, nsr_id
, vld
):
3824 self
.log
.debug("NSR {} create VL {}".format(nsr_id
, vld
))
3825 nsr
= self
._nsrs
[nsr_id
]
3826 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3827 raise NsrVlUpdateError("Cannot perform VL instantiate if NSR is not in running state")
3829 # Not calling in a separate task as this is called from a separate task
3830 yield from nsr
.create_vl_instance(vld
)
3832 def nsr_terminate_vl(self
, nsr_id
, vld
):
3833 self
.log
.debug("NSR {} delete VL {}".format(nsr_id
, vld
.id))
3834 nsr
= self
._nsrs
[nsr_id
]
3835 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3836 raise NsrVlUpdateError("Cannot perform VL terminate if NSR is not in running state")
3838 # Not calling in a separate task as this is called from a separate task
3839 yield from nsr
.delete_vl_instance(vld
)
3841 def create_nsr(self
, nsr_msg
, key_pairs
=None,restart_mode
=False):
3842 """ Create an NSR instance """
3843 if nsr_msg
.id in self
._nsrs
:
3844 msg
= "NSR id %s already exists" % nsr_msg
.id
3845 self
._log
.error(msg
)
3846 raise NetworkServiceRecordError(msg
)
3848 self
._log
.info("Create NetworkServiceRecord nsr id %s from nsd_id %s",
3852 nsm_plugin
= self
._ro
_plugin
_selector
.ro_plugin
3853 sdn_account_name
= self
._cloud
_account
_handler
.get_cloud_account_sdn_name(nsr_msg
.cloud_account
)
3855 nsr
= NetworkServiceRecord(self
._dts
,
3863 restart_mode
=restart_mode
,
3864 vlr_handler
=self
._ro
_plugin
_selector
._records
_publisher
._vlr
_pub
_hdlr
3866 self
._nsrs
[nsr_msg
.id] = nsr
3867 nsm_plugin
.create_nsr(nsr_msg
, nsr_msg
.nsd
, key_pairs
)
3871 def delete_nsr(self
, nsr_id
):
3873 Delete NSR with the passed nsr id
3875 del self
._nsrs
[nsr_id
]
3878 def instantiate_ns(self
, nsr_id
, config_xact
):
3879 """ Instantiate an NS instance """
3880 self
._log
.debug("Instantiating Network service id %s", nsr_id
)
3881 if nsr_id
not in self
._nsrs
:
3882 err
= "NSR id %s not found " % nsr_id
3883 self
._log
.error(err
)
3884 raise NetworkServiceRecordError(err
)
3886 nsr
= self
._nsrs
[nsr_id
]
3887 yield from nsr
.nsm_plugin
.instantiate_ns(nsr
, config_xact
)
3890 def update_vnfr(self
, vnfr
):
3891 """Create/Update an VNFR """
3893 vnfr_state
= self
._vnfrs
[vnfr
.id].state
3894 self
._log
.debug("Updating VNFR with state %s: vnfr %s", vnfr_state
, vnfr
)
3896 yield from self
._vnfrs
[vnfr
.id].update_state(vnfr
)
3897 nsr
= self
.find_nsr_for_vnfr(vnfr
.id)
3898 yield from nsr
.update_state()
3900 def find_nsr_for_vnfr(self
, vnfr_id
):
3901 """ Find the NSR which )has the passed vnfr id"""
3902 for nsr
in list(self
.nsrs
.values()):
3903 for vnfr
in list(nsr
.vnfrs
.values()):
3904 if vnfr
.id == vnfr_id
:
3908 def delete_vnfr(self
, vnfr_id
):
3909 """ Delete VNFR with the passed id"""
3910 del self
._vnfrs
[vnfr_id
]
3912 def get_nsd_ref(self
, nsd_id
):
3913 """ Get network service descriptor for the passed nsd_id
3915 nsd
= self
.get_nsd(nsd_id
)
3920 def get_nsr_config(self
, nsd_id
):
3921 xpath
= "C,/nsr:ns-instance-config"
3922 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3924 for result
in results
:
3925 entry
= yield from result
3926 ns_instance_config
= entry
.result
3928 for nsr
in ns_instance_config
.nsr
:
3929 if nsr
.nsd
.id == nsd_id
:
3935 def nsd_unref_by_nsr_id(self
, nsr_id
):
3936 """ Unref the network service descriptor based on NSR id """
3937 self
._log
.debug("NSR Unref called for Nsr Id:%s", nsr_id
)
3938 if nsr_id
in self
._nsrs
:
3939 nsr
= self
._nsrs
[nsr_id
]
3942 nsd
= self
.get_nsd(nsr
.nsd_id
)
3943 self
._log
.debug("Releasing ref on NSD %s held by NSR %s - Curr %d",
3944 nsd
.id, nsr
.id, nsd
.ref_count
)
3946 except NetworkServiceDescriptorError
:
3947 # We store a copy of NSD in NSR and the NSD in nsd-catalog
3952 self
._log
.error("Cannot find NSR with id %s", nsr_id
)
3953 raise NetworkServiceDescriptorUnrefError("No NSR with id" % nsr_id
)
3956 def nsd_unref(self
, nsd_id
):
3957 """ Unref the network service descriptor associated with the id """
3958 nsd
= self
.get_nsd(nsd_id
)
3961 def get_nsd(self
, nsd_id
):
3962 """ Get network service descriptor for the passed nsd_id"""
3963 if nsd_id
not in self
._nsds
:
3964 self
._log
.error("Cannot find NSD id:%s", nsd_id
)
3965 raise NetworkServiceDescriptorError("Cannot find NSD id:%s", nsd_id
)
3967 return self
._nsds
[nsd_id
]
3969 def create_nsd(self
, nsd_msg
):
3970 """ Create a network service descriptor """
3971 self
._log
.debug("Create network service descriptor - %s", nsd_msg
)
3972 if nsd_msg
.id in self
._nsds
:
3973 self
._log
.error("Cannot create NSD %s -NSD ID already exists", nsd_msg
)
3974 raise NetworkServiceDescriptorError("NSD already exists-%s", nsd_msg
.id)
3976 nsd
= NetworkServiceDescriptor(
3983 self
._nsds
[nsd_msg
.id] = nsd
3987 def update_nsd(self
, nsd
):
3988 """ update the Network service descriptor """
3989 self
._log
.debug("Update network service descriptor - %s", nsd
)
3990 if nsd
.id not in self
._nsds
:
3991 self
._log
.debug("No NSD found - creating NSD id = %s", nsd
.id)
3992 self
.create_nsd(nsd
)
3994 self
._log
.debug("Updating NSD id = %s, nsd = %s", nsd
.id, nsd
)
3995 self
._nsds
[nsd
.id].update(nsd
)
3997 def delete_nsd(self
, nsd_id
):
3998 """ Delete the Network service descriptor with the passed id """
3999 self
._log
.debug("Deleting the network service descriptor - %s", nsd_id
)
4000 if nsd_id
not in self
._nsds
:
4001 self
._log
.debug("Delete NSD failed - cannot find nsd-id %s", nsd_id
)
4002 raise NetworkServiceDescriptorNotFound("Cannot find %s", nsd_id
)
4004 if nsd_id
not in self
._nsds
:
4005 self
._log
.debug("Cannot delete NSD id %s reference exists %s",
4007 self
._nsds
[nsd_id
].ref_count
)
4008 raise NetworkServiceDescriptorRefCountExists(
4009 "Cannot delete :%s, ref_count:%s",
4011 self
._nsds
[nsd_id
].ref_count
)
4013 del self
._nsds
[nsd_id
]
4015 def get_vnfd_config(self
, xact
):
4016 vnfd_dts_reg
= self
._vnfd
_dts
_handler
.regh
4017 for cfg
in vnfd_dts_reg
.get_xact_elements(xact
):
4018 if cfg
.id not in self
._vnfds
:
4019 self
.create_vnfd(cfg
)
4021 def get_vnfd(self
, vnfd_id
, xact
):
4022 """ Get virtual network function descriptor for the passed vnfd_id"""
4023 if vnfd_id
not in self
._vnfds
:
4024 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4025 self
.get_vnfd_config(xact
)
4027 if vnfd_id
not in self
._vnfds
:
4028 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4029 raise VnfDescriptorError("Cannot find VNFD id:%s", vnfd_id
)
4031 return self
._vnfds
[vnfd_id
]
4033 def create_vnfd(self
, vnfd
):
4034 """ Create a virtual network function descriptor """
4035 self
._log
.debug("Create virtual network function descriptor - %s", vnfd
)
4036 if vnfd
.id in self
._vnfds
:
4037 self
._log
.error("Cannot create VNFD %s -VNFD ID already exists", vnfd
)
4038 raise VnfDescriptorError("VNFD already exists-%s", vnfd
.id)
4040 self
._vnfds
[vnfd
.id] = vnfd
4041 return self
._vnfds
[vnfd
.id]
4043 def update_vnfd(self
, vnfd
):
4044 """ Update the virtual network function descriptor """
4045 self
._log
.debug("Update virtual network function descriptor- %s", vnfd
)
4048 if vnfd
.id not in self
._vnfds
:
4049 self
._log
.debug("No VNFD found - creating VNFD id = %s", vnfd
.id)
4050 self
.create_vnfd(vnfd
)
4052 self
._log
.debug("Updating VNFD id = %s, vnfd = %s", vnfd
.id, vnfd
)
4053 self
._vnfds
[vnfd
.id] = vnfd
4056 def delete_vnfd(self
, vnfd_id
):
4057 """ Delete the virtual network function descriptor with the passed id """
4058 self
._log
.debug("Deleting the virtual network function descriptor - %s", vnfd_id
)
4059 if vnfd_id
not in self
._vnfds
:
4060 self
._log
.debug("Delete VNFD failed - cannot find vnfd-id %s", vnfd_id
)
4061 raise VnfDescriptorError("Cannot find %s", vnfd_id
)
4063 del self
._vnfds
[vnfd_id
]
4065 def nsd_in_use(self
, nsd_id
):
4066 """ Is the NSD with the passed id in use """
4067 self
._log
.debug("Is this NSD in use - msg:%s", nsd_id
)
4068 if nsd_id
in self
._nsds
:
4069 return self
._nsds
[nsd_id
].in_use()
4073 def publish_nsr(self
, xact
, path
, msg
):
4074 """ Publish a NSR """
4075 self
._log
.debug("Publish NSR with path %s, msg %s",
4077 yield from self
.nsr_handler
.update(xact
, path
, msg
)
4080 def unpublish_nsr(self
, xact
, path
):
4081 """ Un Publish an NSR """
4082 self
._log
.debug("Publishing delete NSR with path %s", path
)
4083 yield from self
.nsr_handler
.delete(path
, xact
)
4085 def vnfr_is_ready(self
, vnfr_id
):
4086 """ VNFR with the id is ready """
4087 self
._log
.debug("VNFR id %s ready", vnfr_id
)
4088 if vnfr_id
not in self
._vnfds
:
4089 err
= "Did not find VNFR ID with id %s" % vnfr_id
4090 self
._log
.critical("err")
4091 raise VirtualNetworkFunctionRecordError(err
)
4092 self
._vnfrs
[vnfr_id
].is_ready()
4095 def get_nsd_refcount(self
, nsd_id
):
4096 """ Get the nsd_list from this NSM"""
4098 def nsd_refcount_xpath(nsd_id
):
4099 """ xpath for ref count entry """
4100 return (NsdRefCountDtsHandler
.XPATH
+
4101 "[rw-nsr:nsd-id-ref = '{}']").format(nsd_id
)
4104 if nsd_id
is None or nsd_id
== "":
4105 for nsd
in self
._nsds
.values():
4106 nsd_msg
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4107 nsd_msg
.nsd_id_ref
= nsd
.id
4108 nsd_msg
.instance_ref_count
= nsd
.ref_count
4109 nsd_list
.append((nsd_refcount_xpath(nsd
.id), nsd_msg
))
4110 elif nsd_id
in self
._nsds
:
4111 nsd_msg
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4112 nsd_msg
.nsd_id_ref
= self
._nsds
[nsd_id
].id
4113 nsd_msg
.instance_ref_count
= self
._nsds
[nsd_id
].ref_count
4114 nsd_list
.append((nsd_refcount_xpath(nsd_id
), nsd_msg
))
4119 def terminate_ns(self
, nsr_id
, xact
):
4121 Terminate network service for the given NSR Id
4124 # Terminate the instances/networks assocaited with this nw service
4125 self
._log
.debug("Terminating the network service %s", nsr_id
)
4127 yield from self
._nsrs
[nsr_id
].terminate()
4128 except Exception as e
:
4129 self
.log
.exception("Failed to terminate NSR[id=%s]", nsr_id
)
4132 yield from self
.nsd_unref_by_nsr_id(nsr_id
)
4134 # Unpublish the NSR record
4135 self
._log
.debug("Unpublishing the network service %s", nsr_id
)
4136 yield from self
._nsrs
[nsr_id
].unpublish(xact
)
4138 # Finaly delete the NS instance from this NS Manager
4139 self
._log
.debug("Deletng the network service %s", nsr_id
)
4140 self
.delete_nsr(nsr_id
)
4143 class NsmRecordsPublisherProxy(object):
4144 """ This class provides a publisher interface that allows plugin objects
4145 to publish NSR/VNFR/VLR"""
4147 def __init__(self
, dts
, log
, loop
, nsr_pub_hdlr
, vnfr_pub_hdlr
, vlr_pub_hdlr
):
4151 self
._nsr
_pub
_hdlr
= nsr_pub_hdlr
4152 self
._vlr
_pub
_hdlr
= vlr_pub_hdlr
4153 self
._vnfr
_pub
_hdlr
= vnfr_pub_hdlr
4156 def publish_nsr(self
, xact
, nsr
):
4157 """ Publish an NSR """
4158 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4159 return (yield from self
._nsr
_pub
_hdlr
.update(xact
, path
, nsr
))
4162 def unpublish_nsr(self
, xact
, nsr
):
4163 """ Unpublish an NSR """
4164 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4165 return (yield from self
._nsr
_pub
_hdlr
.delete(xact
, path
))
4168 def publish_vnfr(self
, xact
, vnfr
):
4169 """ Publish an VNFR """
4170 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4171 return (yield from self
._vnfr
_pub
_hdlr
.update(xact
, path
, vnfr
))
4174 def unpublish_vnfr(self
, xact
, vnfr
):
4175 """ Unpublish a VNFR """
4176 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4177 return (yield from self
._vnfr
_pub
_hdlr
.delete(xact
, path
))
4180 def publish_vlr(self
, xact
, vlr
):
4181 """ Publish a VLR """
4182 path
= VirtualLinkRecord
.vlr_xpath(vlr
)
4183 return (yield from self
._vlr
_pub
_hdlr
.update(xact
, path
, vlr
))
4186 def unpublish_vlr(self
, xact
, vlr
):
4187 """ Unpublish a VLR """
4188 path
= VirtualLinkRecord
.vlr_xpath(vlr
)
4189 return (yield from self
._vlr
_pub
_hdlr
.delete(xact
, path
))
4192 class ScalingRpcHandler(mano_dts
.DtsHandler
):
4193 """ The Network service Monitor DTS handler """
4194 SCALE_IN_INPUT_XPATH
= "I,/nsr:exec-scale-in"
4195 SCALE_IN_OUTPUT_XPATH
= "O,/nsr:exec-scale-in"
4197 SCALE_OUT_INPUT_XPATH
= "I,/nsr:exec-scale-out"
4198 SCALE_OUT_OUTPUT_XPATH
= "O,/nsr:exec-scale-out"
4200 ACTION
= Enum('ACTION', 'SCALE_IN SCALE_OUT')
4202 def __init__(self
, log
, dts
, loop
, callback
=None):
4203 super().__init
__(log
, dts
, loop
)
4204 self
.callback
= callback
4205 self
.last_instance_id
= defaultdict(int)
4211 def on_scale_in_prepare(xact_info
, action
, ks_path
, msg
):
4212 assert action
== rwdts
.QueryAction
.RPC
4216 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_IN
)
4218 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleIn
.from_dict({
4219 "instance_id": msg
.instance_id
})
4221 xact_info
.respond_xpath(
4222 rwdts
.XactRspCode
.ACK
,
4223 self
.__class
__.SCALE_IN_OUTPUT_XPATH
,
4226 except Exception as e
:
4227 self
.log
.exception(e
)
4228 xact_info
.respond_xpath(
4229 rwdts
.XactRspCode
.NACK
,
4230 self
.__class
__.SCALE_IN_OUTPUT_XPATH
)
4233 def on_scale_out_prepare(xact_info
, action
, ks_path
, msg
):
4234 assert action
== rwdts
.QueryAction
.RPC
4237 scaling_group
= msg
.scaling_group_name_ref
4238 if not msg
.instance_id
:
4239 last_instance_id
= self
.last_instance_id
[scale_group
]
4240 msg
.instance_id
= last_instance_id
+ 1
4241 self
.last_instance_id
[scale_group
] += 1
4244 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_OUT
)
4246 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleOut
.from_dict({
4247 "instance_id": msg
.instance_id
})
4249 xact_info
.respond_xpath(
4250 rwdts
.XactRspCode
.ACK
,
4251 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
,
4254 except Exception as e
:
4255 self
.log
.exception(e
)
4256 xact_info
.respond_xpath(
4257 rwdts
.XactRspCode
.NACK
,
4258 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
)
4260 scale_in_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4261 on_prepare
=on_scale_in_prepare
)
4262 scale_out_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4263 on_prepare
=on_scale_out_prepare
)
4265 with self
.dts
.group_create() as group
:
4267 xpath
=self
.__class
__.SCALE_IN_INPUT_XPATH
,
4268 handler
=scale_in_hdl
,
4269 flags
=rwdts
.Flag
.PUBLISHER
)
4271 xpath
=self
.__class
__.SCALE_OUT_INPUT_XPATH
,
4272 handler
=scale_out_hdl
,
4273 flags
=rwdts
.Flag
.PUBLISHER
)
4276 class NsmTasklet(rift
.tasklets
.Tasklet
):
4278 The network service manager tasklet
4280 def __init__(self
, *args
, **kwargs
):
4281 super(NsmTasklet
, self
).__init
__(*args
, **kwargs
)
4282 self
.rwlog
.set_category("rw-mano-log")
4283 self
.rwlog
.set_subcategory("nsm")
4288 self
._ro
_plugin
_selector
= None
4289 self
._vnffgmgr
= None
4291 self
._nsr
_handler
= None
4292 self
._vnfr
_pub
_handler
= None
4293 self
._vlr
_pub
_handler
= None
4294 self
._vnfd
_pub
_handler
= None
4295 self
._scale
_cfg
_handler
= None
4297 self
._records
_publisher
_proxy
= None
4300 """ The task start callback """
4301 super(NsmTasklet
, self
).start()
4302 self
.log
.info("Starting NsmTasklet")
4304 self
.log
.debug("Registering with dts")
4305 self
._dts
= rift
.tasklets
.DTS(self
.tasklet_info
,
4306 RwNsmYang
.get_schema(),
4308 self
.on_dts_state_change
)
4310 self
.log
.debug("Created DTS Api GI Object: %s", self
._dts
)
4316 print("Caught Exception in NSM stop:", sys
.exc_info()[0])
4319 def on_instance_started(self
):
4320 """ Task instance started callback """
4321 self
.log
.debug("Got instance started callback")
4325 """ Task init callback """
4326 self
.log
.debug("Got instance started callback")
4328 self
.log
.debug("creating config account handler")
4330 self
._nsr
_pub
_handler
= publisher
.NsrOpDataDtsHandler(self
._dts
, self
.log
, self
.loop
)
4331 yield from self
._nsr
_pub
_handler
.register()
4333 self
._vnfr
_pub
_handler
= publisher
.VnfrPublisherDtsHandler(self
._dts
, self
.log
, self
.loop
)
4334 yield from self
._vnfr
_pub
_handler
.register()
4336 self
._vlr
_pub
_handler
= publisher
.VlrPublisherDtsHandler(self
._dts
, self
.log
, self
.loop
)
4337 yield from self
._vlr
_pub
_handler
.register()
4339 manifest
= self
.tasklet_info
.get_pb_manifest()
4340 use_ssl
= manifest
.bootstrap_phase
.rwsecurity
.use_ssl
4341 ssl_cert
= manifest
.bootstrap_phase
.rwsecurity
.cert
4342 ssl_key
= manifest
.bootstrap_phase
.rwsecurity
.key
4344 self
._vnfd
_pub
_handler
= publisher
.VnfdPublisher(use_ssl
, ssl_cert
, ssl_key
, self
.loop
)
4346 self
._records
_publisher
_proxy
= NsmRecordsPublisherProxy(
4350 self
._nsr
_pub
_handler
,
4351 self
._vnfr
_pub
_handler
,
4352 self
._vlr
_pub
_handler
,
4355 # Register the NSM to receive the nsm plugin
4356 # when cloud account is configured
4357 self
._ro
_plugin
_selector
= cloud
.ROAccountPluginSelector(
4361 self
._records
_publisher
_proxy
,
4363 yield from self
._ro
_plugin
_selector
.register()
4365 self
._cloud
_account
_handler
= cloud
.CloudAccountConfigSubscriber(
4370 yield from self
._cloud
_account
_handler
.register()
4372 self
._vnffgmgr
= rwvnffgmgr
.VnffgMgr(self
._dts
, self
.log
, self
.log_hdl
, self
.loop
)
4373 yield from self
._vnffgmgr
.register()
4375 self
._nsm
= NsManager(
4379 self
._nsr
_pub
_handler
,
4380 self
._vnfr
_pub
_handler
,
4381 self
._vlr
_pub
_handler
,
4382 self
._ro
_plugin
_selector
,
4384 self
._vnfd
_pub
_handler
,
4385 self
._cloud
_account
_handler
4388 yield from self
._nsm
.register()
4392 """ Task run callback """
4396 def on_dts_state_change(self
, state
):
4397 """Take action according to current dts state to transition
4398 application into the corresponding application state
4401 state - current dts state
4404 rwdts
.State
.INIT
: rwdts
.State
.REGN_COMPLETE
,
4405 rwdts
.State
.CONFIG
: rwdts
.State
.RUN
,
4409 rwdts
.State
.INIT
: self
.init
,
4410 rwdts
.State
.RUN
: self
.run
,
4413 # Transition application to next state
4414 handler
= handlers
.get(state
, None)
4415 if handler
is not None:
4416 yield from handler()
4418 # Transition dts to next state
4419 next_state
= switch
.get(state
, None)
4420 if next_state
is not None:
4421 self
.log
.debug("Changing state to %s", next_state
)
4422 self
._dts
.handle
.set_state(next_state
)