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('NsdYang', '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
= RwsdnalYang
.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
)
1299 self
._nsm
_plugin
.set_state(self
.id, state
)
1303 """ Get id for this NSR"""
1304 return self
._nsr
_cfg
_msg
.id
1308 """ Name of this network service record """
1309 return self
._nsr
_cfg
_msg
.name
1312 def cloud_account_name(self
):
1313 return self
._nsr
_cfg
_msg
.cloud_account
1316 def om_datacenter_name(self
):
1317 if self
._nsr
_cfg
_msg
.has_field('om_datacenter'):
1318 return self
._nsr
_cfg
_msg
.om_datacenter
1323 """State of this NetworkServiceRecord"""
1324 return self
._op
_status
.state
1328 """ Is this NSR active ?"""
1329 return True if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
else False
1333 """ VLRs associated with this NSR"""
1338 """ VNFRs associated with this NSR"""
1343 """ VNFFGRs associated with this NSR"""
1344 return self
._vnffgrs
1347 def scaling_groups(self
):
1348 """ Scaling groups associated with this NSR """
1349 return self
._scaling
_groups
1352 def param_pools(self
):
1353 """ Parameter value pools associated with this NSR"""
1354 return self
._param
_pools
1357 def nsr_cfg_msg(self
):
1358 return self
._nsr
_cfg
_msg
1361 def nsr_cfg_msg(self
, msg
):
1362 self
._nsr
_cfg
_msg
= msg
1366 """ NSD Protobuf for this NSR """
1367 if self
._nsd
is not None:
1369 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
1374 """ NSD ID for this NSR """
1375 return self
.nsd_msg
.id
1379 ''' Get a new job id for config primitive'''
1384 def config_status(self
):
1385 """ Config status for NSR """
1386 return self
._config
_status
1388 def resolve_placement_group_cloud_construct(self
, input_group
):
1390 Returns the cloud specific construct for placement group
1392 copy_dict
= ['name', 'requirement', 'strategy']
1394 for group_info
in self
._nsr
_cfg
_msg
.nsd_placement_group_maps
:
1395 if group_info
.placement_group_ref
== input_group
.name
:
1396 group
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_PlacementGroupsInfo()
1397 group_dict
= {k
:v
for k
,v
in
1398 group_info
.as_dict().items() if k
!= 'placement_group_ref'}
1399 for param
in copy_dict
:
1400 group_dict
.update({param
: getattr(input_group
, param
)})
1401 group
.from_dict(group_dict
)
1407 return "NSR(name={}, nsd_id={}, cloud_account={})".format(
1408 self
.name
, self
.nsd_id
, self
.cloud_account_name
1411 def _get_vnfd(self
, vnfd_id
, config_xact
):
1412 """ Fetch vnfd msg for the passed vnfd id """
1413 return self
._nsm
.get_vnfd(vnfd_id
, config_xact
)
1415 def _get_vnfd_cloud_account(self
, vnfd_member_index
):
1416 """ Fetch Cloud Account for the passed vnfd id """
1417 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1418 vim_accounts
= [(vnf
.cloud_account
,vnf
.om_datacenter
) for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map \
1419 if vnfd_member_index
== vnf
.member_vnf_index_ref
]
1420 if vim_accounts
and vim_accounts
[0]:
1421 return vim_accounts
[0]
1422 return (self
.cloud_account_name
,self
.om_datacenter_name
)
1424 def _get_constituent_vnfd_msg(self
, vnf_index
):
1425 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1426 if const_vnfd
.member_vnf_index
== vnf_index
:
1429 raise ValueError("Constituent VNF index %s not found" % vnf_index
)
1431 def record_event(self
, evt
, evt_desc
, evt_details
=None, state
=None):
1432 """ Record an event """
1433 self
._op
_status
.record_event(evt
, evt_desc
, evt_details
)
1434 if state
is not None:
1435 self
.set_state(state
)
1437 def scaling_trigger_str(self
, trigger
):
1438 SCALING_TRIGGER_STRS
= {
1439 NsdYang
.ScalingTrigger
.PRE_SCALE_IN
: 'pre-scale-in',
1440 NsdYang
.ScalingTrigger
.POST_SCALE_IN
: 'post-scale-in',
1441 NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
: 'pre-scale-out',
1442 NsdYang
.ScalingTrigger
.POST_SCALE_OUT
: 'post-scale-out',
1445 return SCALING_TRIGGER_STRS
[trigger
]
1446 except Exception as e
:
1447 self
._log
.error("Scaling trigger mapping error for {} : {}".
1449 self
._log
.exception(e
)
1450 return "Unknown trigger"
1453 def instantiate_vls(self
):
1455 This function instantiates VLs for every VL in this Network Service
1457 self
._log
.debug("Instantiating %d VLs in NSD id %s", len(self
._vlrs
),
1459 for vlr
in self
._vlrs
:
1460 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1461 vlr
.state
= VlRecordState
.ACTIVE
1462 self
.vlr_uptime_tasks
[vlr
.id] = self
._loop
.create_task(self
.vlr_uptime_update(vlr
))
1465 def vlr_uptime_update(self
, vlr
):
1468 vlr_
= RwVlrYang
.YangData_Vlr_VlrCatalog_Vlr
.from_dict({'id': vlr
.id})
1470 vlr_
.uptime
= int(time
.time()) - vlr
._create
_time
1471 yield from self
._vlr
_handler
.update(None, VirtualLinkRecord
.vlr_xpath(vlr
), vlr_
)
1472 yield from asyncio
.sleep(2, loop
=self
._loop
)
1473 except asyncio
.CancelledError
:
1474 self
._log
.debug("Received cancellation request for vlr_uptime_update task")
1475 yield from self
._vlr
_handler
.delete(None, VirtualLinkRecord
.vlr_xpath(vlr
))
1479 def create(self
, config_xact
):
1480 """ Create this network service"""
1481 # Create virtual links for all the external vnf
1482 # connection points in this NS
1483 yield from self
.create_vls()
1485 # Create VNFs in this network service
1486 yield from self
.create_vnfs(config_xact
)
1488 # Create VNFFG for network service
1489 self
.create_vnffgs()
1491 # Create Scaling Groups for each scaling group in NSD
1492 self
.create_scaling_groups()
1494 # Create Parameter Pools
1495 self
.create_param_pools()
1498 def apply_scale_group_config_script(self
, script
, group
, scale_instance
, trigger
, vnfrs
=None):
1499 """ Apply config based on script for scale group """
1502 def add_vnfrs_data(vnfrs_list
):
1503 """ Add as a dict each of the VNFRs data """
1505 for vnfr
in vnfrs_list
:
1506 self
._log
.debug("Add VNFR {} data".format(vnfr
))
1508 vnfr_data
['name'] = vnfr
.name
1509 if trigger
in [NsdYang
.ScalingTrigger
.PRE_SCALE_IN
, NsdYang
.ScalingTrigger
.POST_SCALE_OUT
]:
1510 # Get VNF management and other IPs, etc
1511 opdata
= yield from self
.fetch_vnfr(vnfr
.xpath
)
1512 self
._log
.debug("VNFR {} op data: {}".format(vnfr
.name
, opdata
))
1514 vnfr_data
['rw_mgmt_ip'] = opdata
.mgmt_interface
.ip_address
1515 vnfr_data
['rw_mgmt_port'] = opdata
.mgmt_interface
.port
1516 except Exception as e
:
1517 self
._log
.error("Unable to get management IP for vnfr {}:{}".
1518 format(vnfr
.name
, e
))
1521 vnfr_data
['connection_points'] = []
1522 for cp
in opdata
.connection_point
:
1524 con_pt
['name'] = cp
.name
1525 con_pt
['ip_address'] = cp
.ip_address
1526 vnfr_data
['connection_points'].append(con_pt
)
1527 except Exception as e
:
1528 self
._log
.error("Exception getting connections points for VNFR {}: {}".
1529 format(vnfr
.name
, e
))
1531 vnfrs_data
.append(vnfr_data
)
1532 self
._log
.debug("VNFRs data: {}".format(vnfrs_data
))
1536 def add_nsr_data(nsr
):
1538 nsr_data
['name'] = nsr
.name
1541 if script
is None or len(script
) == 0:
1542 self
._log
.error("Script not provided for scale group config: {}".format(group
.name
))
1545 if script
[0] == '/':
1548 path
= os
.path
.join(os
.environ
['RIFT_INSTALL'], "usr/bin", script
)
1549 if not os
.path
.exists(path
):
1550 self
._log
.error("Config faled for scale group {}: Script does not exist at {}".
1551 format(group
.name
, path
))
1554 # Build a YAML file with all parameters for the script to execute
1555 # The data consists of 5 sections
1557 # 2. Scale group config
1558 # 3. VNFRs in the scale group
1559 # 4. VNFRs outside scale group
1562 data
['trigger'] = group
.trigger_map(trigger
)
1563 data
['config'] = group
.group_msg
.as_dict()
1566 data
["vnfrs_in_group"] = yield from add_vnfrs_data(vnfrs
)
1568 data
["vnfrs_in_group"] = yield from add_vnfrs_data(scale_instance
.vnfrs
)
1570 data
["vnfrs_others"] = yield from add_vnfrs_data(self
.vnfrs
.values())
1571 data
["nsr"] = add_nsr_data(self
)
1574 with tempfile
.NamedTemporaryFile(delete
=False) as tmp_file
:
1575 tmp_file
.write(yaml
.dump(data
, default_flow_style
=True)
1578 self
._log
.debug("Creating a temp file: {} with input data: {}".
1579 format(tmp_file
.name
, data
))
1581 cmd
= "{} {}".format(path
, tmp_file
.name
)
1582 self
._log
.debug("Running the CMD: {}".format(cmd
))
1583 proc
= yield from asyncio
.create_subprocess_shell(cmd
, loop
=self
._loop
)
1584 rc
= yield from proc
.wait()
1586 self
._log
.error("The script {} for scale group {} config returned: {}".
1587 format(script
, group
.name
, rc
))
1595 def apply_scaling_group_config(self
, trigger
, group
, scale_instance
, vnfrs
=None):
1596 """ Apply the config for the scaling group based on trigger """
1597 if group
is None or scale_instance
is None:
1601 def update_config_status(success
=True, err_msg
=None):
1602 self
._log
.debug("Update %s config status to %r : %s",
1603 scale_instance
, success
, err_msg
)
1604 if (scale_instance
.config_status
== "failed"):
1605 # Do not update the config status if it is already in failed state
1608 if scale_instance
.config_status
== "configured":
1609 # Update only to failed state an already configured scale instance
1611 scale_instance
.config_status
= "failed"
1612 scale_instance
.config_err_msg
= err_msg
1613 yield from self
.update_state()
1615 # We are in configuring state
1616 # Only after post scale out mark instance as configured
1617 if trigger
== NsdYang
.ScalingTrigger
.POST_SCALE_OUT
:
1619 scale_instance
.config_status
= "configured"
1621 scale_instance
.config_status
= "failed"
1622 scale_instance
.config_err_msg
= err_msg
1623 yield from self
.update_state()
1625 config
= group
.trigger_config(trigger
)
1629 self
._log
.debug("Scaling group {} config: {}".format(group
.name
, config
))
1630 if config
.has_field("ns_config_primitive_name_ref"):
1631 config_name
= config
.ns_config_primitive_name_ref
1632 nsd_msg
= self
.nsd_msg
1633 config_primitive
= None
1634 for ns_cfg_prim
in nsd_msg
.service_primitive
:
1635 if ns_cfg_prim
.name
== config_name
:
1636 config_primitive
= ns_cfg_prim
1639 if config_primitive
is None:
1640 raise ValueError("Could not find ns_cfg_prim %s in nsr %s" % (config_name
, self
.name
))
1642 self
._log
.debug("Scaling group {} config primitive: {}".format(group
.name
, config_primitive
))
1643 if config_primitive
.has_field("user_defined_script"):
1644 rc
= yield from self
.apply_scale_group_config_script(config_primitive
.user_defined_script
,
1645 group
, scale_instance
, trigger
, vnfrs
)
1648 err_msg
= "Failed config for trigger {} using config script '{}'". \
1649 format(self
.scaling_trigger_str(trigger
),
1650 config_primitive
.user_defined_script
)
1651 yield from update_config_status(success
=rc
, err_msg
=err_msg
)
1654 err_msg
= "Failed config for trigger {} as config script is not specified". \
1655 format(self
.scaling_trigger_str(trigger
))
1656 yield from update_config_status(success
=False, err_msg
=err_msg
)
1657 raise NotImplementedError("Only script based config support for scale group for now: {}".
1660 err_msg
= "Failed config for trigger {} as config primitive is not specified".\
1661 format(self
.scaling_trigger_str(trigger
))
1662 yield from update_config_status(success
=False, err_msg
=err_msg
)
1663 self
._log
.error("Config primitive not specified for config action in scale group %s" %
1667 def create_scaling_groups(self
):
1668 """ This function creates a NSScalingGroup for every scaling
1669 group defined in he NSD"""
1671 for scaling_group_msg
in self
.nsd_msg
.scaling_group_descriptor
:
1672 self
._log
.debug("Found scaling_group %s in nsr id %s",
1673 scaling_group_msg
.name
, self
.id)
1675 group_record
= scale_group
.ScalingGroup(
1680 self
._scaling
_groups
[group_record
.name
] = group_record
1683 def create_scale_group_instance(self
, group_name
, index
, config_xact
, is_default
=False):
1684 group
= self
._scaling
_groups
[group_name
]
1685 scale_instance
= group
.create_instance(index
, is_default
)
1689 self
._log
.debug("Creating %u VNFs associated with NS id %s scaling group %s",
1690 len(self
.nsd_msg
.constituent_vnfd
), self
.id, self
)
1693 for vnf_index
, count
in group
.vnf_index_count_map
.items():
1694 const_vnfd_msg
= self
._get
_constituent
_vnfd
_msg
(vnf_index
)
1695 vnfd_msg
= self
._get
_vnfd
(const_vnfd_msg
.vnfd_id_ref
, config_xact
)
1697 cloud_account_name
, om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd_msg
.member_vnf_index
)
1698 if cloud_account_name
is None:
1699 cloud_account_name
= self
.cloud_account_name
1700 for _
in range(count
):
1701 vnfr
= yield from self
.create_vnf_record(vnfd_msg
, const_vnfd_msg
, cloud_account_name
, om_datacenter_name
, group_name
, index
)
1702 scale_instance
.add_vnfr(vnfr
)
1708 def instantiate_instance():
1709 self
._log
.debug("Creating %s VNFRS", scale_instance
)
1710 vnfrs
= yield from create_vnfs()
1711 yield from self
.publish()
1713 self
._log
.debug("Instantiating %s VNFRS for %s", len(vnfrs
), scale_instance
)
1714 scale_instance
.operational_status
= "vnf_init_phase"
1715 yield from self
.update_state()
1718 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
,
1719 group
, scale_instance
, vnfrs
)
1721 self
._log
.error("Pre scale out config for scale group {} ({}) failed".
1722 format(group
.name
, index
))
1723 scale_instance
.operational_status
= "failed"
1725 yield from self
.instantiate_vnfs(vnfrs
)
1727 except Exception as e
:
1728 self
._log
.exception("Failed to begin instantiatiation of vnfs for scale group {}: {}".
1729 format(group
.name
, e
))
1730 self
._log
.exception(e
)
1731 scale_instance
.operational_status
= "failed"
1733 yield from self
.update_state()
1735 yield from instantiate_instance()
1738 def delete_scale_group_instance(self
, group_name
, index
):
1739 group
= self
._scaling
_groups
[group_name
]
1740 scale_instance
= group
.get_instance(index
)
1741 if scale_instance
.is_default
:
1742 raise ScalingOperationError("Cannot terminate a default scaling group instance")
1744 scale_instance
.operational_status
= "terminate"
1745 yield from self
.update_state()
1748 def terminate_instance():
1749 self
._log
.debug("Terminating %s VNFRS" % scale_instance
)
1750 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_IN
,
1751 group
, scale_instance
)
1753 self
._log
.error("Pre scale in config for scale group {} ({}) failed".
1754 format(group
.name
, index
))
1756 # Going ahead with terminate, even if there is an error in pre-scale-in config
1757 # as this could be result of scale out failure and we need to cleanup this group
1758 yield from self
.terminate_vnfrs(scale_instance
.vnfrs
)
1759 group
.delete_instance(index
)
1761 scale_instance
.operational_status
= "vnf_terminate_phase"
1762 yield from self
.update_state()
1764 yield from terminate_instance()
1767 def _update_scale_group_instances_status(self
):
1769 def post_scale_out_task(group
, instance
):
1770 # Apply post scale out config once all VNFRs are active
1771 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_OUT
,
1773 instance
.operational_status
= "running"
1775 self
._log
.debug("Scale out for group {} and instance {} succeeded".
1776 format(group
.name
, instance
.instance_id
))
1778 self
._log
.error("Post scale out config for scale group {} ({}) failed".
1779 format(group
.name
, instance
.instance_id
))
1781 yield from self
.update_state()
1783 group_instances
= {group
: group
.instances
for group
in self
._scaling
_groups
.values()}
1784 for group
, instances
in group_instances
.items():
1785 self
._log
.debug("Updating %s instance status", group
)
1786 for instance
in instances
:
1787 instance_vnf_state_list
= [vnfr
.state
for vnfr
in instance
.vnfrs
]
1788 self
._log
.debug("Got vnfr instance states: %s", instance_vnf_state_list
)
1789 if instance
.operational_status
== "vnf_init_phase":
1790 if all([state
== VnfRecordState
.ACTIVE
for state
in instance_vnf_state_list
]):
1791 instance
.operational_status
= "running"
1793 # Create a task for post scale out to allow us to sleep before attempting
1794 # to configure newly created VM's
1795 self
._loop
.create_task(post_scale_out_task(group
, instance
))
1797 elif any([state
== VnfRecordState
.FAILED
for state
in instance_vnf_state_list
]):
1798 self
._log
.debug("Scale out for group {} and instance {} failed".
1799 format(group
.name
, instance
.instance_id
))
1800 instance
.operational_status
= "failed"
1802 elif instance
.operational_status
== "vnf_terminate_phase":
1803 if all([state
== VnfRecordState
.TERMINATED
for state
in instance_vnf_state_list
]):
1804 instance
.operational_status
= "terminated"
1805 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_IN
,
1808 self
._log
.debug("Scale in for group {} and instance {} succeeded".
1809 format(group
.name
, instance
.instance_id
))
1811 self
._log
.error("Post scale in config for scale group {} ({}) failed".
1812 format(group
.name
, instance
.instance_id
))
1814 def create_vnffgs(self
):
1815 """ This function creates VNFFGs for every VNFFG in the NSD
1816 associated with this NSR"""
1818 for vnffgd
in self
.nsd_msg
.vnffgd
:
1819 self
._log
.debug("Found vnffgd %s in nsr id %s", vnffgd
, self
.id)
1820 vnffgr
= VnffgRecord(self
._dts
,
1823 self
._nsm
._vnffgmgr
,
1827 self
._sdn
_account
_name
1829 self
._vnffgrs
[vnffgr
.id] = vnffgr
1831 def resolve_vld_ip_profile(self
, nsd_msg
, vld
):
1832 self
._log
.debug("Receieved ip profile ref is %s",vld
.ip_profile_ref
)
1833 if not vld
.has_field('ip_profile_ref'):
1835 profile
= [profile
for profile
in nsd_msg
.ip_profiles
if profile
.name
== vld
.ip_profile_ref
]
1836 return profile
[0] if profile
else None
1839 def _create_vls(self
, vld
, cloud_account
,om_datacenter
):
1840 """Create a VLR in the cloud account specified using the given VLD
1844 cloud_account : Cloud account name
1849 vlr
= yield from VirtualLinkRecord
.create_record(
1857 self
.resolve_vld_ip_profile(self
.nsd_msg
, vld
),
1859 restart_mode
=self
.restart_mode
)
1863 def _extract_cloud_accounts_for_vl(self
, vld
):
1865 Extracts the list of cloud accounts from the NS Config obj
1868 1. Cloud accounts based connection point (vnf_cloud_account_map)
1870 vld : VLD yang object
1875 cloud_account_list
= []
1877 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1878 # Handle case where cloud_account is None
1880 for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1881 if vnf
.cloud_account
is not None or vnf
.om_datacenter
is not None:
1882 vnf_cloud_map
[vnf
.member_vnf_index_ref
] = (vnf
.cloud_account
,vnf
.om_datacenter
)
1884 for vnfc
in vld
.vnfd_connection_point_ref
:
1885 cloud_account
= vnf_cloud_map
.get(
1886 vnfc
.member_vnf_index_ref
,
1887 (self
.cloud_account_name
,self
.om_datacenter_name
))
1889 cloud_account_list
.append(cloud_account
)
1891 if self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1892 for vld_map
in self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1893 if vld_map
.vld_id_ref
== vld
.id:
1894 for cloud_account
in vld_map
.cloud_accounts
:
1895 cloud_account_list
.extend((cloud_account
,None))
1896 for om_datacenter
in vld_map
.om_datacenters
:
1897 cloud_account_list
.extend((None,om_datacenter
))
1899 # If no config has been provided then fall-back to the default
1901 if not cloud_account_list
:
1902 cloud_account_list
= [(self
.cloud_account_name
,self
.om_datacenter_name
)]
1904 self
._log
.debug("VL {} cloud accounts: {}".
1905 format(vld
.name
, cloud_account_list
))
1906 return set(cloud_account_list
)
1909 def create_vls(self
):
1910 """ This function creates VLs for every VLD in the NSD
1911 associated with this NSR"""
1912 for vld
in self
.nsd_msg
.vld
:
1914 self
._log
.debug("Found vld %s in nsr id %s", vld
, self
.id)
1915 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1916 for cloud_account
,om_datacenter
in cloud_account_list
:
1917 vlr
= yield from self
._create
_vls
(vld
, cloud_account
,om_datacenter
)
1918 self
._vlrs
.append(vlr
)
1922 def create_vl_instance(self
, vld
):
1923 self
._log
.debug("Create VL for {}: {}".format(self
.id, vld
.as_dict()))
1924 # Check if the VL is already present
1926 for vl
in self
._vlrs
:
1927 if vl
.vld_msg
.id == vld
.id:
1928 self
._log
.debug("The VLD %s already in NSR %s as VLR %s with status %s",
1929 vld
.id, self
.id, vl
.id, vl
.state
)
1931 if vlr
.state
!= VlRecordState
.TERMINATED
:
1932 err_msg
= "VLR for VL %s in NSR %s already instantiated", \
1934 self
._log
.error(err_msg
)
1935 raise NsrVlUpdateError(err_msg
)
1939 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1940 for account
,om_datacenter
in cloud_account_list
:
1941 vlr
= yield from self
._create
_vls
(vld
, account
,om_datacenter
)
1942 self
._vlrs
.append(vlr
)
1944 vlr
.state
= VlRecordState
.INSTANTIATION_PENDING
1945 yield from self
.update_state()
1948 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1949 vlr
.state
= VlRecordState
.ACTIVE
1951 except Exception as e
:
1952 err_msg
= "Error instantiating VL for NSR {} and VLD {}: {}". \
1953 format(self
.id, vld
.id, e
)
1954 self
._log
.error(err_msg
)
1955 self
._log
.exception(e
)
1956 vlr
.state
= VlRecordState
.FAILED
1958 yield from self
.update_state()
1961 def delete_vl_instance(self
, vld
):
1962 for vlr
in self
._vlrs
:
1963 if vlr
.vld_msg
.id == vld
.id:
1964 self
._log
.debug("Found VLR %s for VLD %s in NSR %s",
1965 vlr
.id, vld
.id, self
.id)
1966 vlr
.state
= VlRecordState
.TERMINATE_PENDING
1967 yield from self
.update_state()
1970 yield from self
.nsm_plugin
.terminate_vl(vlr
)
1971 vlr
.state
= VlRecordState
.TERMINATED
1972 self
._vlrs
.remove(vlr
)
1974 except Exception as e
:
1975 err_msg
= "Error terminating VL for NSR {} and VLD {}: {}". \
1976 format(self
.id, vld
.id, e
)
1977 self
._log
.error(err_msg
)
1978 self
._log
.exception(e
)
1979 vlr
.state
= VlRecordState
.FAILED
1981 yield from self
.update_state()
1985 def create_vnfs(self
, config_xact
):
1987 This function creates VNFs for every VNF in the NSD
1988 associated with this NSR
1990 self
._log
.debug("Creating %u VNFs associated with this NS id %s",
1991 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
1993 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1994 if not const_vnfd
.start_by_default
:
1995 self
._log
.debug("start_by_default set to False in constituent VNF (%s). Skipping start.",
1996 const_vnfd
.member_vnf_index
)
1999 vnfd_msg
= self
._get
_vnfd
(const_vnfd
.vnfd_id_ref
, config_xact
)
2000 cloud_account_name
,om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd
.member_vnf_index
)
2001 if cloud_account_name
is None:
2002 cloud_account_name
= self
.cloud_account_name
2003 yield from self
.create_vnf_record(vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
)
2006 def get_placement_groups(self
, vnfd_msg
, const_vnfd
):
2007 placement_groups
= []
2008 for group
in self
.nsd_msg
.placement_groups
:
2009 for member_vnfd
in group
.member_vnfd
:
2010 if (member_vnfd
.vnfd_id_ref
== vnfd_msg
.id) and \
2011 (member_vnfd
.member_vnf_index_ref
== const_vnfd
.member_vnf_index
):
2012 group_info
= self
.resolve_placement_group_cloud_construct(group
)
2013 if group_info
is None:
2014 self
._log
.info("Could not resolve cloud-construct for placement group: %s", group
.name
)
2015 ### raise PlacementGroupError("Could not resolve cloud-construct for placement group: {}".format(group.name))
2017 self
._log
.info("Successfully resolved cloud construct for placement group: %s for VNF: %s (Member Index: %s)",
2020 const_vnfd
.member_vnf_index
)
2021 placement_groups
.append(group_info
)
2022 return placement_groups
2025 def create_vnf_record(self
, vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
, group_name
=None, group_instance_id
=None):
2026 # Fetch the VNFD associated with this VNF
2027 placement_groups
= self
.get_placement_groups(vnfd_msg
, const_vnfd
)
2028 self
._log
.info("Cloud Account for VNF %d is %s",const_vnfd
.member_vnf_index
,cloud_account_name
)
2029 self
._log
.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s, restart mode self.restart_mode %s",
2031 const_vnfd
.member_vnf_index
,
2032 [ group
.name
for group
in placement_groups
],
2034 vnfr
= yield from VirtualNetworkFunctionRecord
.create_record(self
._dts
,
2047 restart_mode
=self
.restart_mode
,
2049 if vnfr
.id in self
._vnfrs
:
2050 err
= "VNF with VNFR id %s already in vnf list" % (vnfr
.id,)
2051 raise NetworkServiceRecordError(err
)
2053 self
._vnfrs
[vnfr
.id] = vnfr
2054 self
._nsm
.vnfrs
[vnfr
.id] = vnfr
2056 yield from vnfr
.set_config_status(NsrYang
.ConfigStates
.INIT
)
2058 self
._log
.debug("Added VNFR %s to NSM VNFR list with id %s",
2064 def create_param_pools(self
):
2065 for param_pool
in self
.nsd_msg
.parameter_pool
:
2066 self
._log
.debug("Found parameter pool %s in nsr id %s", param_pool
, self
.id)
2068 start_value
= param_pool
.range.start_value
2069 end_value
= param_pool
.range.end_value
2070 if end_value
< start_value
:
2071 raise NetworkServiceRecordError(
2072 "Parameter pool %s has invalid range (start: {}, end: {})".format(
2073 start_value
, end_value
2077 self
._param
_pools
[param_pool
.name
] = config_value_pool
.ParameterValuePool(
2080 range(start_value
, end_value
)
2084 def fetch_vnfr(self
, vnfr_path
):
2085 """ Fetch VNFR record """
2087 self
._log
.debug("Fetching VNFR with key %s while instantiating %s",
2089 res_iter
= yield from self
._dts
.query_read(vnfr_path
, rwdts
.XactFlag
.MERGE
)
2091 for ent
in res_iter
:
2092 res
= yield from ent
2098 def instantiate_vnfs(self
, vnfrs
):
2100 This function instantiates VNFs for every VNF in this Network Service
2102 self
._log
.debug("Instantiating %u VNFs in NS %s", len(vnfrs
), self
.id)
2104 self
._log
.debug("Instantiating VNF: %s in NS %s", vnf
, self
.id)
2105 yield from self
.nsm_plugin
.instantiate_vnf(self
, vnf
)
2108 def instantiate_vnffgs(self
):
2110 This function instantiates VNFFGs for every VNFFG in this Network Service
2112 self
._log
.debug("Instantiating %u VNFFGs in NS %s",
2113 len(self
.nsd_msg
.vnffgd
), self
.id)
2114 for _
, vnfr
in self
.vnfrs
.items():
2115 while vnfr
.state
in [VnfRecordState
.INSTANTIATION_PENDING
, VnfRecordState
.INIT
]:
2116 self
._log
.debug("Received vnfr state for vnfr %s is %s; retrying",vnfr
.name
,vnfr
.state
)
2117 yield from asyncio
.sleep(2, loop
=self
._loop
)
2118 if vnfr
.state
== VnfRecordState
.ACTIVE
:
2119 self
._log
.debug("Received vnfr state for vnfr %s is %s ",vnfr
.name
,vnfr
.state
)
2122 self
._log
.debug("Received vnfr state for vnfr %s is %s; failing vnffg creation",vnfr
.name
,vnfr
.state
)
2123 self
._vnffgr
_state
= VnffgRecordState
.FAILED
2126 self
._log
.info("Waiting for 90 seconds for VMs to come up")
2127 yield from asyncio
.sleep(90, loop
=self
._loop
)
2128 self
._log
.info("Starting VNFFG orchestration")
2129 for vnffg
in self
._vnffgrs
.values():
2130 self
._log
.debug("Instantiating VNFFG: %s in NS %s", vnffg
, self
.id)
2131 yield from vnffg
.instantiate()
2134 def instantiate_scaling_instances(self
, config_xact
):
2135 """ Instantiate any default scaling instances in this Network Service """
2136 for group
in self
._scaling
_groups
.values():
2137 for i
in range(group
.min_instance_count
):
2138 self
._log
.debug("Instantiating %s default scaling instance %s", group
, i
)
2139 yield from self
.create_scale_group_instance(
2140 group
.name
, i
, config_xact
, is_default
=True
2143 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2144 if group_msg
.scaling_group_name_ref
!= group
.name
:
2147 for instance
in group_msg
.instance
:
2148 self
._log
.debug("Reloading %s scaling instance %s", group_msg
, instance
.id)
2149 yield from self
.create_scale_group_instance(
2150 group
.name
, instance
.id, config_xact
, is_default
=False
2153 def has_scaling_instances(self
):
2154 """ Return boolean indicating if the network service has default scaling groups """
2155 for group
in self
._scaling
_groups
.values():
2156 if group
.min_instance_count
> 0:
2159 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2160 if len(group_msg
.instance
) > 0:
2167 """ This function publishes this NSR """
2168 self
._nsr
_msg
= self
.create_msg()
2170 self
._log
.debug("Publishing the NSR with xpath %s and nsr %s",
2174 if self
._debug
_running
:
2175 self
._log
.debug("Publishing NSR in RUNNING state!")
2178 with self
._dts
.transaction() as xact
:
2179 yield from self
._nsm
.nsr_handler
.update(xact
, self
.nsr_xpath
, self
._nsr
_msg
)
2180 if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
:
2181 self
._debug
_running
= True
2184 def unpublish(self
, xact
):
2185 """ Unpublish this NSR object """
2186 self
._log
.debug("Unpublishing Network service id %s", self
.id)
2187 yield from self
._nsm
.nsr_handler
.delete(xact
, self
.nsr_xpath
)
2190 def nsr_xpath(self
):
2191 """ Returns the xpath associated with this NSR """
2193 "D,/nsr:ns-instance-opdata" +
2194 "/nsr:nsr[nsr:ns-instance-config-ref = '{}']"
2198 def xpath_from_nsr(nsr
):
2199 """ Returns the xpath associated with this NSR op data"""
2200 return (NetworkServiceRecord
.XPATH
+
2201 "[nsr:ns-instance-config-ref = '{}']").format(nsr
.id)
2204 def nsd_xpath(self
):
2205 """ Return NSD config xpath."""
2207 "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}']"
2208 ).format(self
.nsd_id
)
2211 def instantiate(self
, config_xact
):
2212 """"Instantiates a NetworkServiceRecord.
2214 This function instantiates a Network service
2215 which involves the following steps,
2217 * Instantiate every VL in NSD by sending create VLR request to DTS.
2218 * Instantiate every VNF in NSD by sending create VNF reuqest to DTS.
2219 * Publish the NSR details to DTS
2222 nsr: The NSR configuration request containing nsr-id and nsd
2223 config_xact: The configuration transaction which initiated the instatiation
2226 NetworkServiceRecordError if the NSR creation fails
2232 self
._log
.debug("Instantiating NS - %s xact - %s", self
, config_xact
)
2234 # Move the state to INIITALIZING
2235 self
.set_state(NetworkServiceRecordState
.INIT
)
2237 event_descr
= "Instantiation Request Received NSR Id:%s" % self
.id
2238 self
.record_event("instantiating", event_descr
)
2241 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
2244 # Update ref count if nsd present in catalog
2245 self
._nsm
.get_nsd_ref(self
.nsd_id
)
2247 except NetworkServiceDescriptorError
:
2248 # This could be an NSD not in the nsd-catalog
2251 # Merge any config and initial config primitive values
2252 self
.config_store
.merge_nsd_config(self
.nsd_msg
)
2253 self
._log
.debug("Merged NSD: {}".format(self
.nsd_msg
.as_dict()))
2255 event_descr
= "Fetched NSD with descriptor id %s" % self
.nsd_id
2256 self
.record_event("nsd-fetched", event_descr
)
2258 if self
._nsd
is None:
2259 msg
= "Failed to fetch NSD with nsd-id [%s] for nsr-id %s"
2260 self
._log
.debug(msg
, self
.nsd_id
, self
.id)
2261 raise NetworkServiceRecordError(self
)
2263 self
._log
.debug("Got nsd result %s", self
._nsd
)
2265 # Substitute any input parameters
2266 self
.substitute_input_parameters(self
._nsd
, self
._nsr
_cfg
_msg
)
2269 yield from self
.create(config_xact
)
2271 # Publish the NSR to DTS
2272 yield from self
.publish()
2275 def do_instantiate():
2277 Instantiate network service
2279 self
._log
.debug("Instantiating VLs nsr id [%s] nsd id [%s]",
2280 self
.id, self
.nsd_id
)
2282 # instantiate the VLs
2283 event_descr
= ("Instantiating %s external VLs for NSR id %s" %
2284 (len(self
.nsd_msg
.vld
), self
.id))
2285 self
.record_event("begin-external-vls-instantiation", event_descr
)
2287 self
.set_state(NetworkServiceRecordState
.VL_INIT_PHASE
)
2289 yield from self
.instantiate_vls()
2291 # Publish the NSR to DTS
2292 yield from self
.publish()
2294 event_descr
= ("Finished instantiating %s external VLs for NSR id %s" %
2295 (len(self
.nsd_msg
.vld
), self
.id))
2296 self
.record_event("end-external-vls-instantiation", event_descr
)
2298 self
.set_state(NetworkServiceRecordState
.VNF_INIT_PHASE
)
2300 self
._log
.debug("Instantiating VNFs ...... nsr[%s], nsd[%s]",
2301 self
.id, self
.nsd_id
)
2303 # instantiate the VNFs
2304 event_descr
= ("Instantiating %s VNFS for NSR id %s" %
2305 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2307 self
.record_event("begin-vnf-instantiation", event_descr
)
2309 yield from self
.instantiate_vnfs(self
._vnfrs
.values())
2311 self
._log
.debug(" Finished instantiating %d VNFs for NSR id %s",
2312 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
2314 event_descr
= ("Finished instantiating %s VNFs for NSR id %s" %
2315 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2316 self
.record_event("end-vnf-instantiation", event_descr
)
2318 if len(self
.vnffgrs
) > 0:
2319 #self.set_state(NetworkServiceRecordState.VNFFG_INIT_PHASE)
2320 event_descr
= ("Instantiating %s VNFFGS for NSR id %s" %
2321 (len(self
.nsd_msg
.vnffgd
), self
.id))
2323 self
.record_event("begin-vnffg-instantiation", event_descr
)
2325 yield from self
.instantiate_vnffgs()
2327 event_descr
= ("Finished instantiating %s VNFFGDs for NSR id %s" %
2328 (len(self
.nsd_msg
.vnffgd
), self
.id))
2329 self
.record_event("end-vnffg-instantiation", event_descr
)
2331 if self
.has_scaling_instances():
2332 event_descr
= ("Instantiating %s Scaling Groups for NSR id %s" %
2333 (len(self
._scaling
_groups
), self
.id))
2335 self
.record_event("begin-scaling-group-instantiation", event_descr
)
2336 yield from self
.instantiate_scaling_instances(config_xact
)
2337 self
.record_event("end-scaling-group-instantiation", event_descr
)
2339 # Give the plugin a chance to deploy the network service now that all
2340 # virtual links and vnfs are instantiated
2341 yield from self
.nsm_plugin
.deploy(self
._nsr
_msg
)
2343 self
._log
.debug("Publishing NSR...... nsr[%s], nsd[%s]",
2344 self
.id, self
.nsd_id
)
2346 # Publish the NSR to DTS
2347 yield from self
.publish()
2349 self
._log
.debug("Published NSR...... nsr[%s], nsd[%s]",
2350 self
.id, self
.nsd_id
)
2352 def on_instantiate_done(fut
):
2353 # If the do_instantiate fails, then publish NSR with failed result
2354 if fut
.exception() is not None:
2355 self
._log
.error("NSR instantiation failed for NSR id %s: %s", self
.id, str(fut
.exception()))
2356 self
._loop
.create_task(self
.instantiation_failed(failed_reason
=str(fut
.exception())))
2358 instantiate_task
= self
._loop
.create_task(do_instantiate())
2359 instantiate_task
.add_done_callback(on_instantiate_done
)
2362 def set_config_status(self
, status
, status_details
=None):
2363 if self
.config_status
!= status
:
2364 self
._log
.debug("Updating NSR {} status for {} to {}".
2365 format(self
.name
, self
.config_status
, status
))
2366 self
._config
_status
= status
2367 self
._config
_status
_details
= status_details
2369 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2370 self
.record_event("config-failed", "NS configuration failed",
2371 evt_details
=self
._config
_status
_details
)
2373 yield from self
.publish()
2376 def is_active(self
):
2377 """ This NS is active """
2378 self
.set_state(NetworkServiceRecordState
.RUNNING
)
2382 # Publish the NSR to DTS
2383 self
._log
.debug("Network service %s is active ", self
.id)
2384 self
._is
_active
= True
2386 event_descr
= "NSR in running state for NSR id %s" % self
.id
2387 self
.record_event("ns-running", event_descr
)
2389 yield from self
.publish()
2392 def instantiation_failed(self
, failed_reason
=None):
2393 """ The NS instantiation failed"""
2394 self
._log
.error("Network service id:%s, name:%s instantiation failed",
2396 self
.set_state(NetworkServiceRecordState
.FAILED
)
2398 event_descr
= "Instantiation of NS %s failed" % self
.id
2399 self
.record_event("ns-failed", event_descr
, evt_details
=failed_reason
)
2401 # Publish the NSR to DTS
2402 yield from self
.publish()
2405 def terminate_vnfrs(self
, vnfrs
):
2406 """ Terminate VNFRS in this network service """
2407 self
._log
.debug("Terminating VNFs in network service %s", self
.id)
2409 yield from self
.nsm_plugin
.terminate_vnf(vnfr
)
2412 def terminate(self
):
2413 """ Terminate a NetworkServiceRecord."""
2414 def terminate_vnffgrs():
2415 """ Terminate VNFFGRS in this network service """
2416 self
._log
.debug("Terminating VNFFGRs in network service %s", self
.id)
2417 for vnffgr
in self
.vnffgrs
.values():
2418 yield from vnffgr
.terminate()
2420 def terminate_vlrs():
2421 """ Terminate VLRs in this netork service """
2422 self
._log
.debug("Terminating VLs in network service %s", self
.id)
2423 for vlr
in self
.vlrs
:
2424 yield from self
.nsm_plugin
.terminate_vl(vlr
)
2425 vlr
.state
= VlRecordState
.TERMINATED
2426 if vlr
.id in self
.vlr_uptime_tasks
:
2427 self
.vlr_uptime_tasks
[vlr
.id].cancel()
2429 self
._log
.debug("Terminating network service id %s", self
.id)
2431 # Move the state to TERMINATE
2432 self
.set_state(NetworkServiceRecordState
.TERMINATE
)
2433 event_descr
= "Terminate being processed for NS Id:%s" % self
.id
2434 self
.record_event("terminate", event_descr
)
2436 # Move the state to VNF_TERMINATE_PHASE
2437 self
._log
.debug("Terminating VNFFGs in NS ID: %s", self
.id)
2438 self
.set_state(NetworkServiceRecordState
.VNFFG_TERMINATE_PHASE
)
2439 event_descr
= "Terminating VNFFGS in NS Id:%s" % self
.id
2440 self
.record_event("terminating-vnffgss", event_descr
)
2441 yield from terminate_vnffgrs()
2443 # Move the state to VNF_TERMINATE_PHASE
2444 self
.set_state(NetworkServiceRecordState
.VNF_TERMINATE_PHASE
)
2445 event_descr
= "Terminating VNFS in NS Id:%s" % self
.id
2446 self
.record_event("terminating-vnfs", event_descr
)
2447 yield from self
.terminate_vnfrs(self
.vnfrs
.values())
2449 # Move the state to VL_TERMINATE_PHASE
2450 self
.set_state(NetworkServiceRecordState
.VL_TERMINATE_PHASE
)
2451 event_descr
= "Terminating VLs in NS Id:%s" % self
.id
2452 self
.record_event("terminating-vls", event_descr
)
2453 yield from terminate_vlrs()
2455 yield from self
.nsm_plugin
.terminate_ns(self
)
2457 # Move the state to TERMINATED
2458 self
.set_state(NetworkServiceRecordState
.TERMINATED
)
2459 event_descr
= "Terminated NS Id:%s" % self
.id
2460 self
.record_event("terminated", event_descr
)
2463 """"Enable a NetworkServiceRecord."""
2467 """"Disable a NetworkServiceRecord."""
2470 def map_config_status(self
):
2471 self
._log
.debug("Config status for ns {} is {}".
2472 format(self
.name
, self
._config
_status
))
2473 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURING
:
2474 return 'configuring'
2475 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2479 def vl_phase_completed(self
):
2480 """ Are VLs created in this NS?"""
2481 return self
._vl
_phase
_completed
2483 def vnf_phase_completed(self
):
2484 """ Are VLs created in this NS?"""
2485 return self
._vnf
_phase
_completed
2487 def create_msg(self
):
2488 """ The network serice record as a message """
2489 nsr_dict
= {"ns_instance_config_ref": self
.id}
2490 nsr
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr
.from_dict(nsr_dict
)
2491 #nsr.cloud_account = self.cloud_account_name
2492 nsr
.sdn_account
= self
._sdn
_account
_name
2493 nsr
.name_ref
= self
.name
2494 nsr
.nsd_ref
= self
.nsd_id
2495 nsr
.nsd_name_ref
= self
.nsd_msg
.name
2496 nsr
.operational_events
= self
._op
_status
.msg
2497 nsr
.operational_status
= self
._op
_status
.yang_str()
2498 nsr
.config_status
= self
.map_config_status()
2499 nsr
.config_status_details
= self
._config
_status
_details
2500 nsr
.create_time
= self
._create
_time
2501 nsr
.uptime
= int(time
.time()) - self
._create
_time
2503 for cfg_prim
in self
.nsd_msg
.service_primitive
:
2504 cfg_prim
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_ServicePrimitive
.from_dict(
2506 nsr
.service_primitive
.append(cfg_prim
)
2508 for init_cfg
in self
.nsd_msg
.initial_config_primitive
:
2509 prim
= NsrYang
.NsrInitialConfigPrimitive
.from_dict(
2511 nsr
.initial_config_primitive
.append(prim
)
2513 if self
.vl_phase_completed():
2514 for vlr
in self
.vlrs
:
2515 nsr
.vlr
.append(vlr
.create_nsr_vlr_msg(self
.vnfrs
.values()))
2517 if self
.vnf_phase_completed():
2518 for vnfr_id
in self
.vnfrs
:
2519 nsr
.constituent_vnfr_ref
.append(self
.vnfrs
[vnfr_id
].const_vnfr_msg
)
2520 for vnffgr
in self
.vnffgrs
.values():
2521 nsr
.vnffgr
.append(vnffgr
.fetch_vnffgr())
2522 for scaling_group
in self
._scaling
_groups
.values():
2523 nsr
.scaling_group_record
.append(scaling_group
.create_record_msg())
2527 def all_vnfs_active(self
):
2528 """ Are all VNFS in this NS active? """
2529 for _
, vnfr
in self
.vnfrs
.items():
2530 if vnfr
.active
is not True:
2535 def update_state(self
):
2536 """ Re-evaluate this NS's state """
2537 curr_state
= self
._op
_status
.state
2539 if curr_state
== NetworkServiceRecordState
.TERMINATED
:
2540 self
._log
.debug("NS (%s) in terminated state, not updating state", self
.id)
2543 new_state
= NetworkServiceRecordState
.RUNNING
2544 self
._log
.info("Received update_state for nsr: %s, curr-state: %s",
2545 self
.id, curr_state
)
2547 # Check all the VNFRs are present
2548 for _
, vnfr
in self
.vnfrs
.items():
2549 if vnfr
.state
in [VnfRecordState
.ACTIVE
, VnfRecordState
.TERMINATED
]:
2551 elif vnfr
.state
== VnfRecordState
.FAILED
:
2552 if vnfr
._prev
_state
!= vnfr
.state
:
2553 event_descr
= "Instantiation of VNF %s failed" % vnfr
.id
2554 event_error_details
= vnfr
.state_failed_reason
2555 self
.record_event("vnf-failed", event_descr
, evt_details
=event_error_details
)
2556 vnfr
.set_state(VnfRecordState
.FAILED
)
2558 self
._log
.info("VNF state did not change, curr=%s, prev=%s",
2559 vnfr
.state
, vnfr
._prev
_state
)
2560 new_state
= NetworkServiceRecordState
.FAILED
2563 self
._log
.info("VNF %s in NSR %s is still not active; current state is: %s",
2564 vnfr
.id, self
.id, vnfr
.state
)
2565 new_state
= curr_state
2567 # If new state is RUNNING; check all VLs
2568 if new_state
== NetworkServiceRecordState
.RUNNING
:
2569 for vl
in self
.vlrs
:
2571 if vl
.state
in [VlRecordState
.ACTIVE
, VlRecordState
.TERMINATED
]:
2573 elif vl
.state
== VlRecordState
.FAILED
:
2574 if vl
.prev_state
!= vl
.state
:
2575 event_descr
= "Instantiation of VL %s failed" % vl
.id
2576 event_error_details
= vl
.state_failed_reason
2577 self
.record_event("vl-failed", event_descr
, evt_details
=event_error_details
)
2578 vl
.prev_state
= vl
.state
2580 self
._log
.debug("VL %s already in failed state")
2582 if vl
.state
in [VlRecordState
.INSTANTIATION_PENDING
, VlRecordState
.INIT
]:
2583 new_state
= NetworkServiceRecordState
.VL_INSTANTIATE
2586 if vl
.state
in [VlRecordState
.TERMINATE_PENDING
]:
2587 new_state
= NetworkServiceRecordState
.VL_TERMINATE
2590 # If new state is RUNNING; check VNFFGRs are also active
2591 if new_state
== NetworkServiceRecordState
.RUNNING
:
2592 for _
, vnffgr
in self
.vnffgrs
.items():
2593 self
._log
.info("Checking vnffgr state for nsr %s is: %s",
2594 self
.id, vnffgr
.state
)
2595 if vnffgr
.state
== VnffgRecordState
.ACTIVE
:
2597 elif vnffgr
.state
== VnffgRecordState
.FAILED
:
2598 event_descr
= "Instantiation of VNFFGR %s failed" % vnffgr
.id
2599 self
.record_event("vnffg-failed", event_descr
)
2600 new_state
= NetworkServiceRecordState
.FAILED
2603 self
._log
.info("VNFFGR %s in NSR %s is still not active; current state is: %s",
2604 vnffgr
.id, self
.id, vnffgr
.state
)
2605 new_state
= curr_state
2607 # Update all the scaling group instance operational status to
2608 # reflect the state of all VNFR within that instance
2609 yield from self
._update
_scale
_group
_instances
_status
()
2611 for _
, group
in self
._scaling
_groups
.items():
2612 if group
.state
== scale_group
.ScaleGroupState
.SCALING_OUT
:
2613 new_state
= NetworkServiceRecordState
.SCALING_OUT
2615 elif group
.state
== scale_group
.ScaleGroupState
.SCALING_IN
:
2616 new_state
= NetworkServiceRecordState
.SCALING_IN
2619 if new_state
!= curr_state
:
2620 self
._log
.debug("Changing state of Network service %s from %s to %s",
2621 self
.id, curr_state
, new_state
)
2622 if new_state
== NetworkServiceRecordState
.RUNNING
:
2623 yield from self
.is_active()
2624 elif new_state
== NetworkServiceRecordState
.FAILED
:
2625 # If the NS is already active and we entered scaling_in, scaling_out,
2626 # do not mark the NS as failing if scaling operation failed.
2627 if curr_state
in [NetworkServiceRecordState
.SCALING_OUT
,
2628 NetworkServiceRecordState
.SCALING_IN
] and self
._is
_active
:
2629 new_state
= NetworkServiceRecordState
.RUNNING
2630 self
.set_state(new_state
)
2632 yield from self
.instantiation_failed()
2634 self
.set_state(new_state
)
2636 yield from self
.publish()
2639 class InputParameterSubstitution(object):
2641 This class is responsible for substituting input parameters into an NSD.
2644 def __init__(self
, log
):
2645 """Create an instance of InputParameterSubstitution
2648 log - a logger for this object to use
2653 def __call__(self
, nsd
, nsr_config
):
2654 """Substitutes input parameters from the NSR config into the NSD
2656 This call modifies the provided NSD with the input parameters that are
2657 contained in the NSR config.
2660 nsd - a GI NSD object
2661 nsr_config - a GI NSR config object
2664 if nsd
is None or nsr_config
is None:
2667 # Create a lookup of the xpath elements that this descriptor allows
2669 optional_input_parameters
= set()
2670 for input_parameter
in nsd
.input_parameter_xpath
:
2671 optional_input_parameters
.add(input_parameter
.xpath
)
2673 # Apply the input parameters to the descriptor
2674 if nsr_config
.input_parameter
:
2675 for param
in nsr_config
.input_parameter
:
2676 if param
.xpath
not in optional_input_parameters
:
2677 msg
= "tried to set an invalid input parameter ({})"
2678 self
.log
.error(msg
.format(param
.xpath
))
2682 "input-parameter:{} = {}".format(
2689 xpath
.setxattr(nsd
, param
.xpath
, param
.value
)
2691 except Exception as e
:
2692 self
.log
.exception(e
)
2695 class NetworkServiceDescriptor(object):
2697 Network service descriptor class
2700 def __init__(self
, dts
, log
, loop
, nsd
, nsm
):
2712 """ Returns nsd id """
2717 """ Returns name of nsd """
2718 return self
._nsd
.name
2721 def ref_count(self
):
2722 """ Returns reference count"""
2723 return self
._ref
_count
2726 """ Returns whether nsd is in use or not """
2727 return True if self
.ref_count
> 0 else False
2730 """ Take a reference on this object """
2731 self
._ref
_count
+= 1
2734 """ Release reference on this object """
2735 if self
.ref_count
< 1:
2736 msg
= ("Unref on a NSD object - nsd id %s, ref_count = %s" %
2737 (self
.id, self
.ref_count
))
2738 self
._log
.critical(msg
)
2739 raise NetworkServiceDescriptorError(msg
)
2740 self
._ref
_count
-= 1
2744 """ Return the message associated with this NetworkServiceDescriptor"""
2748 def path_for_id(nsd_id
):
2749 """ Return path for the passed nsd_id"""
2750 return "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}'".format(nsd_id
)
2753 """ Return the message associated with this NetworkServiceDescriptor"""
2754 return NetworkServiceDescriptor
.path_for_id(self
.id)
2756 def update(self
, nsd
):
2757 """ Update the NSD descriptor """
2761 class NsdDtsHandler(object):
2762 """ The network service descriptor DTS handler """
2763 XPATH
= "C,/nsd:nsd-catalog/nsd:nsd"
2765 def __init__(self
, dts
, log
, loop
, nsm
):
2775 """ Return registration handle """
2780 """ Register for Nsd create/update/delete/read requests from dts """
2782 def on_apply(dts
, acg
, xact
, action
, scratch
):
2783 """Apply the configuration"""
2784 is_recovery
= xact
.xact
is None and action
== rwdts
.AppconfAction
.INSTALL
2785 self
._log
.debug("Got nsd apply cfg (xact:%s) (action:%s)",
2787 # Create/Update an NSD record
2788 for cfg
in self
._regh
.get_xact_elements(xact
):
2789 # Only interested in those NSD cfgs whose ID was received in prepare callback
2790 if cfg
.id in scratch
.get('nsds', []) or is_recovery
:
2791 self
._nsm
.update_nsd(cfg
)
2793 scratch
.pop('nsds', None)
2795 return RwTypes
.RwStatus
.SUCCESS
2798 def delete_nsd_libs(nsd_id
):
2799 """ Remove any files uploaded with NSD and stored under $RIFT_VAR_ROOT/libs/<id> """
2801 rift_var_root_dir
= os
.environ
['RIFT_VAR_ROOT']
2802 nsd_dir
= os
.path
.join(rift_var_root_dir
, 'launchpad/libs', nsd_id
)
2804 if os
.path
.exists (nsd_dir
):
2805 shutil
.rmtree(nsd_dir
, ignore_errors
=True)
2806 except Exception as e
:
2807 self
._log
.error("Exception in cleaning up NSD libs {}: {}".
2809 self
._log
.excpetion(e
)
2812 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2813 """ Prepare callback from DTS for NSD config """
2815 self
._log
.info("Got nsd prepare - config received nsd id %s, msg %s",
2818 fref
= ProtobufC
.FieldReference
.alloc()
2819 fref
.goto_whole_message(msg
.to_pbcm())
2821 if fref
.is_field_deleted():
2822 # Delete an NSD record
2823 self
._log
.debug("Deleting NSD with id %s", msg
.id)
2824 if self
._nsm
.nsd_in_use(msg
.id):
2825 errmsg
= "Cannot delete an NSD in use - %s" % msg
.id
2826 self
._log
.error(errmsg
)
2827 xpath
= ks_path
.to_xpath(NsdYang
.get_schema())
2828 xact_info
.send_error_xpath(RwTypes
.RwStatus
.FAILURE
,
2831 xact_info
.respond_xpath(rwdts
.XactRspCode
.NACK
)
2834 yield from delete_nsd_libs(msg
.id)
2835 self
._nsm
.delete_nsd(msg
.id)
2837 # Add this NSD to scratch to create/update in apply callback
2838 nsds
= scratch
.setdefault('nsds', [])
2840 # acg._scratch['nsds'].append(msg.id)
2842 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2845 "Registering for NSD config using xpath: %s",
2846 NsdDtsHandler
.XPATH
,
2849 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2850 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2851 # Need a list in scratch to store NSDs to create/update later
2852 # acg._scratch['nsds'] = list()
2853 self
._regh
= acg
.register(
2854 xpath
=NsdDtsHandler
.XPATH
,
2855 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
2856 on_prepare
=on_prepare
)
2859 class VnfdDtsHandler(object):
2860 """ DTS handler for VNFD config changes """
2861 XPATH
= "C,/vnfd:vnfd-catalog/vnfd:vnfd"
2863 def __init__(self
, dts
, log
, loop
, nsm
):
2872 """ DTS registration handle """
2877 """ Register for VNFD configuration"""
2880 def on_apply(dts
, acg
, xact
, action
, scratch
):
2881 """Apply the configuration"""
2882 self
._log
.debug("Got NSM VNFD apply (xact: %s) (action: %s)(scr: %s)",
2883 xact
, action
, scratch
)
2885 # Create/Update a VNFD record
2886 for cfg
in self
._regh
.get_xact_elements(xact
):
2887 # Only interested in those VNFD cfgs whose ID was received in prepare callback
2888 if cfg
.id in scratch
.get('vnfds', []):
2889 self
._nsm
.update_vnfd(cfg
)
2891 for cfg
in self
._regh
.elements
:
2892 if cfg
.id in scratch
.get('deleted_vnfds', []):
2893 yield from self
._nsm
.delete_vnfd(cfg
.id)
2895 scratch
.pop('vnfds', None)
2896 scratch
.pop('deleted_vnfds', None)
2899 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2900 """ on prepare callback """
2901 self
._log
.debug("Got on prepare for VNFD (path: %s) (action: %s) (msg: %s)",
2902 ks_path
.to_xpath(RwNsmYang
.get_schema()), xact_info
.query_action
, msg
)
2904 fref
= ProtobufC
.FieldReference
.alloc()
2905 fref
.goto_whole_message(msg
.to_pbcm())
2907 # Handle deletes in prepare_callback, but adds/updates in apply_callback
2908 if fref
.is_field_deleted():
2909 self
._log
.debug("Adding msg to deleted field")
2910 deleted_vnfds
= scratch
.setdefault('deleted_vnfds', [])
2911 deleted_vnfds
.append(msg
.id)
2913 # Add this VNFD to scratch to create/update in apply callback
2914 vnfds
= scratch
.setdefault('vnfds', [])
2915 vnfds
.append(msg
.id)
2917 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2920 "Registering for VNFD config using xpath: %s",
2921 VnfdDtsHandler
.XPATH
,
2923 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2924 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2925 # Need a list in scratch to store VNFDs to create/update later
2926 # acg._scratch['vnfds'] = list()
2927 # acg._scratch['deleted_vnfds'] = list()
2928 self
._regh
= acg
.register(
2929 xpath
=VnfdDtsHandler
.XPATH
,
2930 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY
,
2931 on_prepare
=on_prepare
)
2933 class NsrRpcDtsHandler(object):
2934 """ The network service instantiation RPC DTS handler """
2935 EXEC_NSR_CONF_XPATH
= "I,/nsr:start-network-service"
2936 EXEC_NSR_CONF_O_XPATH
= "O,/nsr:start-network-service"
2937 NETCONF_IP_ADDRESS
= "127.0.0.1"
2939 RESTCONF_PORT
= 8888
2940 NETCONF_USER
= "admin"
2941 NETCONF_PW
= "admin"
2942 REST_BASE_V2_URL
= 'https://{}:{}/v2/api/'.format("127.0.0.1",8888)
2944 def __init__(self
, dts
, log
, loop
, nsm
):
2951 self
._ns
_regh
= None
2953 self
._manager
= None
2954 self
._nsr
_config
_url
= NsrRpcDtsHandler
.REST_BASE_V2_URL
+ 'config/ns-instance-config'
2956 self
._model
= RwYang
.Model
.create_libncx()
2957 self
._model
.load_schema_ypbc(RwNsrYang
.get_schema())
2961 """ Return the NS manager instance """
2965 def wrap_netconf_config_xml(xml
):
2966 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
2970 def _connect(self
, timeout_secs
=240):
2972 start_time
= time
.time()
2973 while (time
.time() - start_time
) < timeout_secs
:
2976 self
._log
.debug("Attemping NsmTasklet netconf connection.")
2978 manager
= yield from ncclient
.asyncio_manager
.asyncio_connect(
2980 host
=NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
,
2981 port
=NsrRpcDtsHandler
.NETCONF_PORT
,
2982 username
=NsrRpcDtsHandler
.NETCONF_USER
,
2983 password
=NsrRpcDtsHandler
.NETCONF_PW
,
2985 look_for_keys
=False,
2986 hostkey_verify
=False,
2991 except ncclient
.transport
.errors
.SSHError
as e
:
2992 self
._log
.warning("Netconf connection to launchpad %s failed: %s",
2993 NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
, str(e
))
2995 yield from asyncio
.sleep(5, loop
=self
._loop
)
2997 raise NsrInstantiationFailed("Failed to connect to Launchpad within %s seconds" %
3000 def _apply_ns_instance_config(self
,payload_dict
):
3001 #self._log.debug("At apply NS instance config with payload %s",payload_dict)
3002 req_hdr
= {'accept':'application/vnd.yang.data+json','content-type':'application/vnd.yang.data+json'}
3003 response
=requests
.post(self
._nsr
_config
_url
, headers
=req_hdr
, auth
=('admin', 'admin'),data
=payload_dict
,verify
=False)
3008 """ Register for NS monitoring read from dts """
3010 def on_ns_config_prepare(xact_info
, action
, ks_path
, msg
):
3011 """ prepare callback from dts start-network-service"""
3012 assert action
== rwdts
.QueryAction
.RPC
3014 rpc_op
= NsrYang
.YangOutput_Nsr_StartNetworkService
.from_dict({
3015 "nsr_id":str(uuid
.uuid4())
3018 if not ('name' in rpc_ip
and 'nsd_ref' in rpc_ip
and
3019 ('cloud_account' in rpc_ip
or 'om_datacenter' in rpc_ip
)):
3021 "Mandatory parameters name or nsd_ref or cloud account not found in start-network-service {}".
3023 self
._log
.error(errmsg
)
3024 xact_info
.send_error_xpath(RwTypes
.RwStatus
.FAILURE
,
3025 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
,
3027 xact_info
.respond_xpath(rwdts
.XactRspCode
.NACK
,
3028 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
)
3031 self
._log
.debug("start-network-service RPC input: {}".format(rpc_ip
))
3034 # Add used value to the pool
3035 self
._log
.debug("RPC output: {}".format(rpc_op
))
3037 nsd_copy
= self
.nsm
.get_nsd(rpc_ip
.nsd_ref
)
3039 self
._log
.debug("Configuring ns-instance-config with name %s nsd-ref: %s",
3040 rpc_ip
.name
, rpc_ip
.nsd_ref
)
3042 ns_instance_config_dict
= {"id":rpc_op
.nsr_id
, "admin_status":"ENABLED"}
3043 ns_instance_config_copy_dict
= {k
:v
for k
, v
in rpc_ip
.as_dict().items()
3044 if k
in RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr().fields
}
3045 ns_instance_config_dict
.update(ns_instance_config_copy_dict
)
3047 ns_instance_config
= RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.from_dict(ns_instance_config_dict
)
3048 ns_instance_config
.nsd
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_Nsd()
3049 ns_instance_config
.nsd
.from_dict(nsd_copy
.msg
.as_dict())
3051 payload_dict
= ns_instance_config
.to_json(self
._model
)
3053 self
._log
.debug("Sending configure ns-instance-config json to %s: %s",
3054 self
._nsr
_config
_url
,ns_instance_config
)
3056 response
= yield from self
._loop
.run_in_executor(
3058 self
._apply
_ns
_instance
_config
,
3061 response
.raise_for_status()
3062 self
._log
.debug("Received edit config response: %s", response
.json())
3064 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
,
3065 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
,
3067 except Exception as e
:
3068 errmsg
= ("Exception processing the "
3069 "start-network-service: {}".format(e
))
3070 self
._log
.exception(errmsg
)
3071 xact_info
.send_error_xpath(RwTypes
.RwStatus
.FAILURE
,
3072 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
,
3074 xact_info
.respond_xpath(rwdts
.XactRspCode
.NACK
,
3075 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
)
3077 hdl_ns
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_ns_config_prepare
,)
3079 with self
._dts
.group_create() as group
:
3080 self
._ns
_regh
= group
.register(xpath
=NsrRpcDtsHandler
.EXEC_NSR_CONF_XPATH
,
3082 flags
=rwdts
.Flag
.PUBLISHER
,
3086 class NsrDtsHandler(object):
3087 """ The network service DTS handler """
3088 NSR_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr"
3089 SCALE_INSTANCE_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr/nsr:scaling-group/nsr:instance"
3090 KEY_PAIR_XPATH
= "C,/nsr:key-pair"
3092 def __init__(self
, dts
, log
, loop
, nsm
):
3098 self
._nsr
_regh
= None
3099 self
._scale
_regh
= None
3100 self
._key
_pair
_regh
= None
3104 """ Return the NS manager instance """
3109 """ Register for Nsr create/update/delete/read requests from dts """
3111 def nsr_id_from_keyspec(ks
):
3112 nsr_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.schema().keyspec_to_entry(ks
)
3113 nsr_id
= nsr_path_entry
.key00
.id
3116 def group_name_from_keyspec(ks
):
3117 group_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
.schema().keyspec_to_entry(ks
)
3118 group_name
= group_path_entry
.key00
.scaling_group_name_ref
3121 def is_instance_in_reg_elements(nsr_id
, group_name
, instance_id
):
3122 """ Return boolean indicating if scaling group instance was already commited previously.
3124 By looking at the existing elements in this registration handle (elements not part
3125 of this current xact), we can tell if the instance was configured previously without
3126 keeping any application state.
3128 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3129 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3130 elem_group_name
= group_name_from_keyspec(keyspec
)
3132 if elem_nsr_id
!= nsr_id
or group_name
!= elem_group_name
:
3135 if instance_cfg
.id == instance_id
:
3140 def get_scale_group_instance_delta(nsr_id
, group_name
, xact
):
3141 delta
= {"added": [], "deleted": []}
3142 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3143 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3144 if elem_nsr_id
!= nsr_id
:
3147 elem_group_name
= group_name_from_keyspec(keyspec
)
3148 if elem_group_name
!= group_name
:
3151 delta
["added"].append(instance_cfg
.id)
3153 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(include_keyspec
=True):
3154 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3155 if elem_nsr_id
!= nsr_id
:
3158 elem_group_name
= group_name_from_keyspec(keyspec
)
3159 if elem_group_name
!= group_name
:
3162 if instance_cfg
.id in delta
["added"]:
3163 delta
["added"].remove(instance_cfg
.id)
3165 delta
["deleted"].append(instance_cfg
.id)
3170 def update_nsr_nsd(nsr_id
, xact
, scratch
):
3173 def get_nsr_vl_delta(nsr_id
, xact
, scratch
):
3174 delta
= {"added": [], "deleted": []}
3175 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3176 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3177 if elem_nsr_id
!= nsr_id
:
3180 if 'vld' in instance_cfg
.nsd
:
3181 for vld
in instance_cfg
.nsd
.vld
:
3182 delta
["added"].append(vld
)
3184 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3185 self
._log
.debug("NSR update: %s", instance_cfg
)
3186 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3187 if elem_nsr_id
!= nsr_id
:
3190 if 'vld' in instance_cfg
.nsd
:
3191 for vld
in instance_cfg
.nsd
.vld
:
3192 if vld
in delta
["added"]:
3193 delta
["added"].remove(vld
)
3195 delta
["deleted"].append(vld
)
3199 vl_delta
= yield from get_nsr_vl_delta(nsr_id
, xact
, scratch
)
3200 self
._log
.debug("Got NSR:%s VL instance delta: %s", nsr_id
, vl_delta
)
3202 for vld
in vl_delta
["added"]:
3203 yield from self
._nsm
.nsr_instantiate_vl(nsr_id
, vld
)
3205 for vld
in vl_delta
["deleted"]:
3206 yield from self
._nsm
.nsr_terminate_vl(nsr_id
, vld
)
3208 def get_add_delete_update_cfgs(dts_member_reg
, xact
, key_name
, scratch
):
3209 # Unfortunately, it is currently difficult to figure out what has exactly
3210 # changed in this xact without Pbdelta support (RIFT-4916)
3211 # As a workaround, we can fetch the pre and post xact elements and
3212 # perform a comparison to figure out adds/deletes/updates
3213 xact_cfgs
= list(dts_member_reg
.get_xact_elements(xact
))
3214 curr_cfgs
= list(dts_member_reg
.elements
)
3216 xact_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in xact_cfgs
}
3217 curr_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in curr_cfgs
}
3220 added_keys
= set(xact_key_map
) - set(curr_key_map
)
3221 added_cfgs
= [xact_key_map
[key
] for key
in added_keys
]
3224 deleted_keys
= set(curr_key_map
) - set(xact_key_map
)
3225 deleted_cfgs
= [curr_key_map
[key
] for key
in deleted_keys
]
3228 updated_keys
= set(curr_key_map
) & set(xact_key_map
)
3229 updated_cfgs
= [xact_key_map
[key
] for key
in updated_keys
3230 if xact_key_map
[key
] != curr_key_map
[key
]]
3232 return added_cfgs
, deleted_cfgs
, updated_cfgs
3234 def get_nsr_key_pairs(dts_member_reg
, xact
):
3236 for instance_cfg
, keyspec
in dts_member_reg
.get_xact_elements(xact
, include_keyspec
=True):
3237 self
._log
.debug("Key pair received is {} KS: {}".format(instance_cfg
, keyspec
))
3238 xpath
= keyspec
.to_xpath(RwNsrYang
.get_schema())
3239 key_pairs
[instance_cfg
.name
] = instance_cfg
3242 def on_apply(dts
, acg
, xact
, action
, scratch
):
3243 """Apply the configuration"""
3244 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3245 xact
, action
, scratch
)
3247 def handle_create_nsr(msg
, key_pairs
=None, restart_mode
=False):
3248 # Handle create nsr requests """
3249 # Do some validations
3250 if not msg
.has_field("nsd"):
3251 err
= "NSD not provided"
3252 self
._log
.error(err
)
3253 raise NetworkServiceRecordError(err
)
3255 self
._log
.debug("Creating NetworkServiceRecord %s from nsr config %s",
3256 msg
.id, msg
.as_dict())
3257 nsr
= self
.nsm
.create_nsr(msg
, key_pairs
=key_pairs
, restart_mode
=restart_mode
)
3260 def handle_delete_nsr(msg
):
3262 def delete_instantiation(ns_id
):
3263 """ Delete instantiation """
3264 with self
._dts
.transaction() as xact
:
3265 yield from self
._nsm
.terminate_ns(ns_id
, xact
)
3267 # Handle delete NSR requests
3268 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3269 # Terminate the NSR instance
3270 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3272 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3273 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3274 nsr
.record_event("terminate-rcvd", event_descr
)
3276 self
._loop
.create_task(delete_instantiation(msg
.id))
3279 def begin_instantiation(nsr
):
3280 # Begin instantiation
3281 self
._log
.info("Beginning NS instantiation: %s", nsr
.id)
3282 yield from self
._nsm
.instantiate_ns(nsr
.id, xact
)
3284 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3285 xact
, action
, scratch
)
3287 if action
== rwdts
.AppconfAction
.INSTALL
and xact
.id is None:
3289 for element
in self
._key
_pair
_regh
.elements
:
3290 key_pairs
.append(element
)
3291 for element
in self
._nsr
_regh
.elements
:
3292 nsr
= handle_create_nsr(element
, key_pairs
, restart_mode
=True)
3293 self
._loop
.create_task(begin_instantiation(nsr
))
3296 (added_msgs
, deleted_msgs
, updated_msgs
) = get_add_delete_update_cfgs(self
._nsr
_regh
,
3300 self
._log
.debug("Added: %s, Deleted: %s, Updated: %s", added_msgs
,
3301 deleted_msgs
, updated_msgs
)
3303 for msg
in added_msgs
:
3304 if msg
.id not in self
._nsm
.nsrs
:
3305 self
._log
.info("Create NSR received in on_apply to instantiate NS:%s", msg
.id)
3306 key_pairs
= get_nsr_key_pairs(self
._key
_pair
_regh
, xact
)
3307 nsr
= handle_create_nsr(msg
,key_pairs
)
3308 self
._loop
.create_task(begin_instantiation(nsr
))
3310 for msg
in deleted_msgs
:
3311 self
._log
.info("Delete NSR received in on_apply to terminate NS:%s", msg
.id)
3313 handle_delete_nsr(msg
)
3315 self
._log
.exception("Failed to terminate NS:%s", msg
.id)
3317 for msg
in updated_msgs
:
3318 self
._log
.info("Update NSR received in on_apply: %s", msg
)
3320 self
._nsm
.nsr_update_cfg(msg
.id, msg
)
3323 self
._loop
.create_task(update_nsr_nsd(msg
.id, xact
, scratch
))
3325 for group
in msg
.scaling_group
:
3326 instance_delta
= get_scale_group_instance_delta(msg
.id, group
.scaling_group_name_ref
, xact
)
3327 self
._log
.debug("Got NSR:%s scale group instance delta: %s", msg
.id, instance_delta
)
3329 for instance_id
in instance_delta
["added"]:
3330 self
._nsm
.scale_nsr_out(msg
.id, group
.scaling_group_name_ref
, instance_id
, xact
)
3332 for instance_id
in instance_delta
["deleted"]:
3333 self
._nsm
.scale_nsr_in(msg
.id, group
.scaling_group_name_ref
, instance_id
)
3336 return RwTypes
.RwStatus
.SUCCESS
3339 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
3340 """ Prepare calllback from DTS for NSR """
3342 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3343 action
= xact_info
.query_action
3345 "Got Nsr prepare callback (xact: %s) (action: %s) (info: %s), %s:%s)",
3346 xact
, action
, xact_info
, xpath
, msg
3350 def delete_instantiation(ns_id
):
3351 """ Delete instantiation """
3352 yield from self
._nsm
.terminate_ns(ns_id
, None)
3354 def handle_delete_nsr():
3355 """ Handle delete NSR requests """
3356 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3357 # Terminate the NSR instance
3358 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3360 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3361 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3362 nsr
.record_event("terminate-rcvd", event_descr
)
3364 self
._loop
.create_task(delete_instantiation(msg
.id))
3366 fref
= ProtobufC
.FieldReference
.alloc()
3367 fref
.goto_whole_message(msg
.to_pbcm())
3369 def send_err_msg(err_msg
):
3370 self
._log
.error(errmsg
)
3371 xact_info
.send_error_xpath(RwTypes
.RwStatus
.FAILURE
,
3374 xact_info
.respond_xpath(rwdts
.XactRspCode
.NACK
)
3377 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
, rwdts
.QueryAction
.DELETE
]:
3378 # if this is an NSR create
3379 if action
!= rwdts
.QueryAction
.DELETE
and msg
.id not in self
._nsm
.nsrs
:
3380 # Ensure the Cloud account/datacenter has been specified
3381 if not msg
.has_field("cloud_account") and not msg
.has_field("om_datacenter"):
3382 errmsg
= ("Cloud account or datacenter not specified in NS {}".
3384 send_err_msg(errmsg
)
3387 # Check if nsd is specified
3388 if not msg
.has_field("nsd"):
3389 errmsg
= ("NSD not specified in NS {}".
3391 send_err_msg(errmsg
)
3395 nsr
= self
._nsm
.nsrs
[msg
.id]
3397 if msg
.has_field("nsd"):
3398 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3399 errmsg
= ("Unable to update VL when NS {} not in running state".
3401 send_err_msg(errmsg
)
3404 if 'vld' not in msg
.nsd
or len(msg
.nsd
.vld
) == 0:
3405 errmsg
= ("NS config {} NSD should have atleast 1 VLD".
3407 send_err_msg(errmsg
)
3410 if msg
.has_field("scaling_group"):
3411 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3412 errmsg
= ("Unable to perform scaling action when NS {} not in running state".
3414 send_err_msg(errmsg
)
3417 if len(msg
.scaling_group
) > 1:
3418 errmsg
= ("Only a single scaling group can be configured at a time for NS {}".
3420 send_err_msg(errmsg
)
3423 for group_msg
in msg
.scaling_group
:
3424 num_new_group_instances
= len(group_msg
.instance
)
3425 if num_new_group_instances
> 1:
3426 errmsg
= ("Only a single scaling instance can be modified at a time for NS {}".
3428 send_err_msg(errmsg
)
3431 elif num_new_group_instances
== 1:
3432 scale_group
= nsr
.scaling_groups
[group_msg
.scaling_group_name_ref
]
3433 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
]:
3434 if len(scale_group
.instances
) == scale_group
.max_instance_count
:
3435 errmsg
= (" Max instances for {} reached for NS {}".
3436 format(str(scale_group
), msg
.name
))
3437 send_err_msg(errmsg
)
3440 acg
.handle
.prepare_complete_ok(xact_info
.handle
)
3443 self
._log
.debug("Registering for NSR config using xpath: %s",
3444 NsrDtsHandler
.NSR_XPATH
)
3446 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
3447 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
3448 self
._nsr
_regh
= acg
.register(xpath
=NsrDtsHandler
.NSR_XPATH
,
3449 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3450 on_prepare
=on_prepare
)
3452 self
._scale
_regh
= acg
.register(
3453 xpath
=NsrDtsHandler
.SCALE_INSTANCE_XPATH
,
3454 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY| rwdts
.Flag
.CACHE
,
3457 self
._key
_pair
_regh
= acg
.register(
3458 xpath
=NsrDtsHandler
.KEY_PAIR_XPATH
,
3459 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3463 class NsrOpDataDtsHandler(object):
3464 """ The network service op data DTS handler """
3465 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
3467 def __init__(self
, dts
, log
, loop
, nsm
):
3476 """ Return the registration handle"""
3481 """ Return the NS manager instance """
3486 """ Register for Nsr op data publisher registration"""
3487 self
._log
.debug("Registering Nsr op data path %s as publisher",
3488 NsrOpDataDtsHandler
.XPATH
)
3490 hdl
= rift
.tasklets
.DTS
.RegistrationHandler()
3491 handlers
= rift
.tasklets
.Group
.Handler()
3492 with self
._dts
.group_create(handler
=handlers
) as group
:
3493 self
._regh
= group
.register(xpath
=NsrOpDataDtsHandler
.XPATH
,
3495 flags
=rwdts
.Flag
.PUBLISHER | rwdts
.Flag
.NO_PREP_READ | rwdts
.Flag
.DATASTORE
)
3498 def create(self
, path
, msg
):
3500 Create an NS record in DTS with the path and message
3502 self
._log
.debug("Creating NSR %s:%s", path
, msg
)
3503 self
.regh
.create_element(path
, msg
)
3504 self
._log
.debug("Created NSR, %s:%s", path
, msg
)
3507 def update(self
, path
, msg
, flags
=rwdts
.XactFlag
.REPLACE
):
3509 Update an NS record in DTS with the path and message
3511 self
._log
.debug("Updating NSR, %s:%s regh = %s", path
, msg
, self
.regh
)
3512 self
.regh
.update_element(path
, msg
, flags
)
3513 self
._log
.debug("Updated NSR, %s:%s", path
, msg
)
3516 def delete(self
, path
):
3518 Update an NS record in DTS with the path and message
3520 self
._log
.debug("Deleting NSR path:%s", path
)
3521 self
.regh
.delete_element(path
)
3522 self
._log
.debug("Deleted NSR path:%s", path
)
3525 class VnfrDtsHandler(object):
3526 """ The virtual network service DTS handler """
3527 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
3529 def __init__(self
, dts
, log
, loop
, nsm
):
3539 """ Return registration handle """
3544 """ Return the NS manager instance """
3549 """ Register for vnfr create/update/delete/ advises from dts """
3551 def on_commit(xact_info
):
3552 """ The transaction has been committed """
3553 self
._log
.debug("Got vnfr commit (xact_info: %s)", xact_info
)
3554 return rwdts
.MemberRspCode
.ACTION_OK
3557 def on_prepare(xact_info
, action
, ks_path
, msg
):
3558 """ prepare callback from dts """
3559 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3561 "Got vnfr on_prepare cb (xact_info: %s, action: %s): %s:%s",
3562 xact_info
, action
, ks_path
, msg
3565 schema
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr
.schema()
3566 path_entry
= schema
.keyspec_to_entry(ks_path
)
3567 if path_entry
.key00
.id not in self
._nsm
._vnfrs
:
3568 self
._log
.error("%s request for non existent record path %s",
3570 xact_info
.respond_xpath(rwdts
.XactRspCode
.NA
, xpath
)
3574 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3575 if action
== rwdts
.QueryAction
.CREATE
or action
== rwdts
.QueryAction
.UPDATE
:
3576 yield from self
._nsm
.update_vnfr(msg
)
3577 elif action
== rwdts
.QueryAction
.DELETE
:
3578 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3579 self
._nsm
.delete_vnfr(path_entry
.key00
.id)
3581 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
, xpath
)
3583 self
._log
.debug("Registering for VNFR using xpath: %s",
3584 VnfrDtsHandler
.XPATH
,)
3586 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_commit
=on_commit
,
3587 on_prepare
=on_prepare
,)
3588 with self
._dts
.group_create() as group
:
3589 self
._regh
= group
.register(xpath
=VnfrDtsHandler
.XPATH
,
3591 flags
=(rwdts
.Flag
.SUBSCRIBER
),)
3594 class NsdRefCountDtsHandler(object):
3595 """ The NSD Ref Count DTS handler """
3596 XPATH
= "D,/nsr:ns-instance-opdata/rw-nsr:nsd-ref-count"
3598 def __init__(self
, dts
, log
, loop
, nsm
):
3608 """ Return registration handle """
3613 """ Return the NS manager instance """
3618 """ Register for NSD ref count read from dts """
3621 def on_prepare(xact_info
, action
, ks_path
, msg
):
3622 """ prepare callback from dts """
3623 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3625 if action
== rwdts
.QueryAction
.READ
:
3626 schema
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount
.schema()
3627 path_entry
= schema
.keyspec_to_entry(ks_path
)
3628 nsd_list
= yield from self
._nsm
.get_nsd_refcount(path_entry
.key00
.nsd_id_ref
)
3629 for xpath
, msg
in nsd_list
:
3630 xact_info
.respond_xpath(rsp_code
=rwdts
.XactRspCode
.MORE
,
3633 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
3635 raise NetworkServiceRecordError("Not supported operation %s" % action
)
3637 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_prepare
,)
3638 with self
._dts
.group_create() as group
:
3639 self
._regh
= group
.register(xpath
=NsdRefCountDtsHandler
.XPATH
,
3641 flags
=rwdts
.Flag
.PUBLISHER
,)
3644 class NsManager(object):
3645 """ The Network Service Manager class"""
3646 def __init__(self
, dts
, log
, loop
,
3647 nsr_handler
, vnfr_handler
, vlr_handler
, ro_plugin_selector
,
3648 vnffgmgr
, vnfd_pub_handler
, cloud_account_handler
):
3652 self
._nsr
_handler
= nsr_handler
3653 self
._vnfr
_pub
_handler
= vnfr_handler
3654 self
._vlr
_pub
_handler
= vlr_handler
3655 self
._vnffgmgr
= vnffgmgr
3656 self
._vnfd
_pub
_handler
= vnfd_pub_handler
3657 self
._cloud
_account
_handler
= cloud_account_handler
3659 self
._ro
_plugin
_selector
= ro_plugin_selector
3661 # Intialize the set of variables for implementing Scaling RPC using REST.
3662 self
._headers
= {"content-type":"application/json", "accept":"application/json"}
3663 #This will break when we have rbac in the rift code and admin user password is changed or admin it self is removed.
3664 self
._user
= 'admin'
3665 self
._password
= 'admin'
3666 self
._ip
= 'localhost'
3668 self
._conf
_url
= "https://{ip}:{port}/api/config". \
3677 self
.cfgmgr_obj
= conman
.ROConfigManager(log
, loop
, dts
, self
)
3679 # TODO: All these handlers should move to tasklet level.
3680 # Passing self is often an indication of bad design
3681 self
._nsd
_dts
_handler
= NsdDtsHandler(dts
, log
, loop
, self
)
3682 self
._vnfd
_dts
_handler
= VnfdDtsHandler(dts
, log
, loop
, self
)
3683 self
._dts
_handlers
= [self
._nsd
_dts
_handler
,
3684 VnfrDtsHandler(dts
, log
, loop
, self
),
3685 NsdRefCountDtsHandler(dts
, log
, loop
, self
),
3686 NsrDtsHandler(dts
, log
, loop
, self
),
3687 ScalingRpcHandler(log
, dts
, loop
, self
.scale_rpc_callback
),
3688 NsrRpcDtsHandler(dts
,log
,loop
,self
),
3689 self
._vnfd
_dts
_handler
,
3710 def nsr_handler(self
):
3711 """" NSR handler """
3712 return self
._nsr
_handler
3716 """" So Obj handler """
3721 """ NSRs in this NSM"""
3726 """ NSDs in this NSM"""
3731 """ VNFDs in this NSM"""
3736 """ VNFRs in this NSM"""
3740 def nsr_pub_handler(self
):
3741 """ NSR publication handler """
3742 return self
._nsr
_handler
3745 def vnfr_pub_handler(self
):
3746 """ VNFR publication handler """
3747 return self
._vnfr
_pub
_handler
3750 def vlr_pub_handler(self
):
3751 """ VLR publication handler """
3752 return self
._vlr
_pub
_handler
3755 def vnfd_pub_handler(self
):
3756 return self
._vnfd
_pub
_handler
3760 """ Register all static DTS handlers """
3761 for dts_handle
in self
._dts
_handlers
:
3762 yield from dts_handle
.register()
3765 def get_ns_by_nsr_id(self
, nsr_id
):
3766 """ get NSR by nsr id """
3767 if nsr_id
not in self
._nsrs
:
3768 raise NetworkServiceRecordError("NSR id %s not found" % nsr_id
)
3770 return self
._nsrs
[nsr_id
]
3772 def scale_nsr_out(self
, nsr_id
, scale_group_name
, instance_id
, config_xact
):
3773 self
.log
.debug("Scale out NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3778 nsr
= self
._nsrs
[nsr_id
]
3779 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3780 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3782 self
._loop
.create_task(nsr
.create_scale_group_instance(scale_group_name
, instance_id
, config_xact
))
3784 def scale_nsr_in(self
, nsr_id
, scale_group_name
, instance_id
):
3785 self
.log
.debug("Scale in NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3790 nsr
= self
._nsrs
[nsr_id
]
3791 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3792 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3794 self
._loop
.create_task(nsr
.delete_scale_group_instance(scale_group_name
, instance_id
))
3796 def scale_rpc_callback(self
, xact
, msg
, action
):
3797 """Callback handler for RPC calls
3799 xact : Transaction Handler
3801 action : Scaling Action
3803 def get_scaling_group_information():
3804 scaling_group_url
= "{url}/ns-instance-config/nsr/{nsr_id}".format(url
=self
._conf
_url
, nsr_id
=msg
.nsr_id_ref
)
3805 output
= requests
.get(scaling_group_url
, headers
=self
._headers
, auth
=(self
._user
, self
._password
), verify
=False)
3806 if output
.text
== None or len(output
.text
) == 0:
3807 self
.log
.error("nsr id %s information not present", self
._nsr
_id
)
3809 scaling_group_info
= json
.loads(output
.text
)
3810 return scaling_group_info
3812 def config_scaling_group_information(scaling_group_info
):
3813 data_str
= json
.dumps(scaling_group_info
)
3814 self
.log
.debug("scaling group Info %s", data_str
)
3816 scale_out_url
= "{url}/ns-instance-config/nsr/{nsr_id}".format(url
=self
._conf
_url
, nsr_id
=msg
.nsr_id_ref
)
3817 response
= requests
.put(scale_out_url
, data
=data_str
, verify
=False, auth
=(self
._user
, self
._password
), headers
=self
._headers
)
3818 response
.raise_for_status()
3821 scaling_group_info
= get_scaling_group_information()
3822 if scaling_group_info
is None:
3825 scaling_group_present
= False
3826 if "scaling-group" in scaling_group_info
["nsr:nsr"]:
3827 scaling_group_array
= scaling_group_info
["nsr:nsr"]["scaling-group"]
3828 for scaling_group
in scaling_group_array
:
3829 if scaling_group
["scaling-group-name-ref"] == msg
.scaling_group_name_ref
:
3830 scaling_group_present
= True
3831 if 'instance' not in scaling_group
:
3832 scaling_group
['instance'] = []
3833 for instance
in scaling_group
['instance']:
3834 if instance
["id"] == int(msg
.instance_id
):
3835 self
.log
.error("scaling group with instance id %s exists for scale out", msg
.instance_id
)
3837 scaling_group
["instance"].append({"id": int(msg
.instance_id
)})
3839 if not scaling_group_present
:
3840 scaling_group_info
["nsr:nsr"]["scaling-group"] = [{"scaling-group-name-ref": msg
.scaling_group_name_ref
, "instance": [{"id": msg
.instance_id
}]}]
3842 config_scaling_group_information(scaling_group_info
)
3846 scaling_group_info
= get_scaling_group_information()
3847 if scaling_group_info
is None:
3850 scaling_group_array
= scaling_group_info
["nsr:nsr"]["scaling-group"]
3851 scaling_group_present
= False
3852 instance_id_present
= False
3853 for scaling_group
in scaling_group_array
:
3854 if scaling_group
["scaling-group-name-ref"] == msg
.scaling_group_name_ref
:
3855 scaling_group_present
= True
3856 if 'instance' in scaling_group
:
3857 instance_array
= scaling_group
["instance"];
3858 for index
in range(len(instance_array
)):
3859 if instance_array
[index
]["id"] == int(msg
.instance_id
):
3860 instance_array
.pop(index
)
3861 instance_id_present
= True
3864 if not scaling_group_present
:
3865 self
.log
.error("Scaling group %s doesnot exists for scale in", msg
.scaling_group_name_ref
)
3868 if not instance_id_present
:
3869 self
.log
.error("Instance id %s doesnot exists for scale in", msg
.instance_id
)
3872 config_scaling_group_information(scaling_group_info
)
3875 if action
== ScalingRpcHandler
.ACTION
.SCALE_OUT
:
3876 self
._loop
.run_in_executor(None, scale_out
)
3878 self
._loop
.run_in_executor(None, scale_in
)
3880 def nsr_update_cfg(self
, nsr_id
, msg
):
3881 nsr
= self
._nsrs
[nsr_id
]
3882 nsr
.nsr_cfg_msg
= msg
3884 def nsr_instantiate_vl(self
, nsr_id
, vld
):
3885 self
.log
.debug("NSR {} create VL {}".format(nsr_id
, vld
))
3886 nsr
= self
._nsrs
[nsr_id
]
3887 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3888 raise NsrVlUpdateError("Cannot perform VL instantiate if NSR is not in running state")
3890 # Not calling in a separate task as this is called from a separate task
3891 yield from nsr
.create_vl_instance(vld
)
3893 def nsr_terminate_vl(self
, nsr_id
, vld
):
3894 self
.log
.debug("NSR {} delete VL {}".format(nsr_id
, vld
.id))
3895 nsr
= self
._nsrs
[nsr_id
]
3896 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3897 raise NsrVlUpdateError("Cannot perform VL terminate if NSR is not in running state")
3899 # Not calling in a separate task as this is called from a separate task
3900 yield from nsr
.delete_vl_instance(vld
)
3902 def create_nsr(self
, nsr_msg
, key_pairs
=None,restart_mode
=False):
3903 """ Create an NSR instance """
3904 if nsr_msg
.id in self
._nsrs
:
3905 msg
= "NSR id %s already exists" % nsr_msg
.id
3906 self
._log
.error(msg
)
3907 raise NetworkServiceRecordError(msg
)
3909 self
._log
.info("Create NetworkServiceRecord nsr id %s from nsd_id %s, restart mode %s",
3914 nsm_plugin
= self
._ro
_plugin
_selector
.ro_plugin
3915 sdn_account_name
= self
._cloud
_account
_handler
.get_cloud_account_sdn_name(nsr_msg
.cloud_account
)
3917 nsr
= NetworkServiceRecord(self
._dts
,
3925 restart_mode
=restart_mode
,
3926 vlr_handler
=self
._ro
_plugin
_selector
._records
_publisher
._vlr
_pub
_hdlr
3928 self
._nsrs
[nsr_msg
.id] = nsr
3929 nsm_plugin
.create_nsr(nsr_msg
, nsr_msg
.nsd
, key_pairs
)
3933 def delete_nsr(self
, nsr_id
):
3935 Delete NSR with the passed nsr id
3937 del self
._nsrs
[nsr_id
]
3940 def instantiate_ns(self
, nsr_id
, config_xact
):
3941 """ Instantiate an NS instance """
3942 self
._log
.debug("Instantiating Network service id %s", nsr_id
)
3943 if nsr_id
not in self
._nsrs
:
3944 err
= "NSR id %s not found " % nsr_id
3945 self
._log
.error(err
)
3946 raise NetworkServiceRecordError(err
)
3948 nsr
= self
._nsrs
[nsr_id
]
3949 yield from nsr
.nsm_plugin
.instantiate_ns(nsr
, config_xact
)
3952 def update_vnfr(self
, vnfr
):
3953 """Create/Update an VNFR """
3955 vnfr_state
= self
._vnfrs
[vnfr
.id].state
3956 self
._log
.debug("Updating VNFR with state %s: vnfr %s", vnfr_state
, vnfr
)
3958 yield from self
._vnfrs
[vnfr
.id].update_state(vnfr
)
3959 nsr
= self
.find_nsr_for_vnfr(vnfr
.id)
3960 yield from nsr
.update_state()
3962 def find_nsr_for_vnfr(self
, vnfr_id
):
3963 """ Find the NSR which )has the passed vnfr id"""
3964 for nsr
in list(self
.nsrs
.values()):
3965 for vnfr
in list(nsr
.vnfrs
.values()):
3966 if vnfr
.id == vnfr_id
:
3970 def delete_vnfr(self
, vnfr_id
):
3971 """ Delete VNFR with the passed id"""
3972 del self
._vnfrs
[vnfr_id
]
3974 def get_nsd_ref(self
, nsd_id
):
3975 """ Get network service descriptor for the passed nsd_id
3977 nsd
= self
.get_nsd(nsd_id
)
3982 def get_nsr_config(self
, nsd_id
):
3983 xpath
= "C,/nsr:ns-instance-config"
3984 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3986 for result
in results
:
3987 entry
= yield from result
3988 ns_instance_config
= entry
.result
3990 for nsr
in ns_instance_config
.nsr
:
3991 if nsr
.nsd
.id == nsd_id
:
3997 def nsd_unref_by_nsr_id(self
, nsr_id
):
3998 """ Unref the network service descriptor based on NSR id """
3999 self
._log
.debug("NSR Unref called for Nsr Id:%s", nsr_id
)
4000 if nsr_id
in self
._nsrs
:
4001 nsr
= self
._nsrs
[nsr_id
]
4004 nsd
= self
.get_nsd(nsr
.nsd_id
)
4005 self
._log
.debug("Releasing ref on NSD %s held by NSR %s - Curr %d",
4006 nsd
.id, nsr
.id, nsd
.ref_count
)
4008 except NetworkServiceDescriptorError
:
4009 # We store a copy of NSD in NSR and the NSD in nsd-catalog
4014 self
._log
.error("Cannot find NSR with id %s", nsr_id
)
4015 raise NetworkServiceDescriptorUnrefError("No NSR with id" % nsr_id
)
4018 def nsd_unref(self
, nsd_id
):
4019 """ Unref the network service descriptor associated with the id """
4020 nsd
= self
.get_nsd(nsd_id
)
4023 def get_nsd(self
, nsd_id
):
4024 """ Get network service descriptor for the passed nsd_id"""
4025 if nsd_id
not in self
._nsds
:
4026 self
._log
.error("Cannot find NSD id:%s", nsd_id
)
4027 raise NetworkServiceDescriptorError("Cannot find NSD id:%s", nsd_id
)
4029 return self
._nsds
[nsd_id
]
4031 def create_nsd(self
, nsd_msg
):
4032 """ Create a network service descriptor """
4033 self
._log
.debug("Create network service descriptor - %s", nsd_msg
)
4034 if nsd_msg
.id in self
._nsds
:
4035 self
._log
.error("Cannot create NSD %s -NSD ID already exists", nsd_msg
)
4036 raise NetworkServiceDescriptorError("NSD already exists-%s", nsd_msg
.id)
4038 nsd
= NetworkServiceDescriptor(
4045 self
._nsds
[nsd_msg
.id] = nsd
4049 def update_nsd(self
, nsd
):
4050 """ update the Network service descriptor """
4051 self
._log
.debug("Update network service descriptor - %s", nsd
)
4052 if nsd
.id not in self
._nsds
:
4053 self
._log
.debug("No NSD found - creating NSD id = %s", nsd
.id)
4054 self
.create_nsd(nsd
)
4056 self
._log
.debug("Updating NSD id = %s, nsd = %s", nsd
.id, nsd
)
4057 self
._nsds
[nsd
.id].update(nsd
)
4059 def delete_nsd(self
, nsd_id
):
4060 """ Delete the Network service descriptor with the passed id """
4061 self
._log
.debug("Deleting the network service descriptor - %s", nsd_id
)
4062 if nsd_id
not in self
._nsds
:
4063 self
._log
.debug("Delete NSD failed - cannot find nsd-id %s", nsd_id
)
4064 raise NetworkServiceDescriptorNotFound("Cannot find %s", nsd_id
)
4066 if nsd_id
not in self
._nsds
:
4067 self
._log
.debug("Cannot delete NSD id %s reference exists %s",
4069 self
._nsds
[nsd_id
].ref_count
)
4070 raise NetworkServiceDescriptorRefCountExists(
4071 "Cannot delete :%s, ref_count:%s",
4073 self
._nsds
[nsd_id
].ref_count
)
4075 del self
._nsds
[nsd_id
]
4077 def get_vnfd_config(self
, xact
):
4078 vnfd_dts_reg
= self
._vnfd
_dts
_handler
.regh
4079 for cfg
in vnfd_dts_reg
.get_xact_elements(xact
):
4080 if cfg
.id not in self
._vnfds
:
4081 self
.create_vnfd(cfg
)
4083 def get_vnfd(self
, vnfd_id
, xact
):
4084 """ Get virtual network function descriptor for the passed vnfd_id"""
4085 if vnfd_id
not in self
._vnfds
:
4086 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4087 self
.get_vnfd_config(xact
)
4089 if vnfd_id
not in self
._vnfds
:
4090 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4091 raise VnfDescriptorError("Cannot find VNFD id:%s", vnfd_id
)
4093 return self
._vnfds
[vnfd_id
]
4095 def create_vnfd(self
, vnfd
):
4096 """ Create a virtual network function descriptor """
4097 self
._log
.debug("Create virtual network function descriptor - %s", vnfd
)
4098 if vnfd
.id in self
._vnfds
:
4099 self
._log
.error("Cannot create VNFD %s -VNFD ID already exists", vnfd
)
4100 raise VnfDescriptorError("VNFD already exists-%s", vnfd
.id)
4102 self
._vnfds
[vnfd
.id] = vnfd
4103 return self
._vnfds
[vnfd
.id]
4105 def update_vnfd(self
, vnfd
):
4106 """ Update the virtual network function descriptor """
4107 self
._log
.debug("Update virtual network function descriptor- %s", vnfd
)
4110 if vnfd
.id not in self
._vnfds
:
4111 self
._log
.debug("No VNFD found - creating VNFD id = %s", vnfd
.id)
4112 self
.create_vnfd(vnfd
)
4114 self
._log
.debug("Updating VNFD id = %s, vnfd = %s", vnfd
.id, vnfd
)
4115 self
._vnfds
[vnfd
.id] = vnfd
4118 def delete_vnfd(self
, vnfd_id
):
4119 """ Delete the virtual network function descriptor with the passed id """
4120 self
._log
.debug("Deleting the virtual network function descriptor - %s", vnfd_id
)
4121 if vnfd_id
not in self
._vnfds
:
4122 self
._log
.debug("Delete VNFD failed - cannot find vnfd-id %s", vnfd_id
)
4123 raise VnfDescriptorError("Cannot find %s", vnfd_id
)
4125 del self
._vnfds
[vnfd_id
]
4127 def nsd_in_use(self
, nsd_id
):
4128 """ Is the NSD with the passed id in use """
4129 self
._log
.debug("Is this NSD in use - msg:%s", nsd_id
)
4130 if nsd_id
in self
._nsds
:
4131 return self
._nsds
[nsd_id
].in_use()
4135 def publish_nsr(self
, xact
, path
, msg
):
4136 """ Publish a NSR """
4137 self
._log
.debug("Publish NSR with path %s, msg %s",
4139 yield from self
.nsr_handler
.update(xact
, path
, msg
)
4142 def unpublish_nsr(self
, xact
, path
):
4143 """ Un Publish an NSR """
4144 self
._log
.debug("Publishing delete NSR with path %s", path
)
4145 yield from self
.nsr_handler
.delete(path
, xact
)
4147 def vnfr_is_ready(self
, vnfr_id
):
4148 """ VNFR with the id is ready """
4149 self
._log
.debug("VNFR id %s ready", vnfr_id
)
4150 if vnfr_id
not in self
._vnfds
:
4151 err
= "Did not find VNFR ID with id %s" % vnfr_id
4152 self
._log
.critical("err")
4153 raise VirtualNetworkFunctionRecordError(err
)
4154 self
._vnfrs
[vnfr_id
].is_ready()
4157 def get_nsd_refcount(self
, nsd_id
):
4158 """ Get the nsd_list from this NSM"""
4160 def nsd_refcount_xpath(nsd_id
):
4161 """ xpath for ref count entry """
4162 return (NsdRefCountDtsHandler
.XPATH
+
4163 "[rw-nsr:nsd-id-ref = '{}']").format(nsd_id
)
4166 if nsd_id
is None or nsd_id
== "":
4167 for nsd
in self
._nsds
.values():
4168 nsd_msg
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4169 nsd_msg
.nsd_id_ref
= nsd
.id
4170 nsd_msg
.instance_ref_count
= nsd
.ref_count
4171 nsd_list
.append((nsd_refcount_xpath(nsd
.id), nsd_msg
))
4172 elif nsd_id
in self
._nsds
:
4173 nsd_msg
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4174 nsd_msg
.nsd_id_ref
= self
._nsds
[nsd_id
].id
4175 nsd_msg
.instance_ref_count
= self
._nsds
[nsd_id
].ref_count
4176 nsd_list
.append((nsd_refcount_xpath(nsd_id
), nsd_msg
))
4181 def terminate_ns(self
, nsr_id
, xact
):
4183 Terminate network service for the given NSR Id
4186 # Terminate the instances/networks assocaited with this nw service
4187 self
._log
.debug("Terminating the network service %s", nsr_id
)
4189 yield from self
._nsrs
[nsr_id
].terminate()
4190 except Exception as e
:
4191 self
.log
.exception("Failed to terminate NSR[id=%s]", nsr_id
)
4194 yield from self
.nsd_unref_by_nsr_id(nsr_id
)
4196 # Unpublish the NSR record
4197 self
._log
.debug("Unpublishing the network service %s", nsr_id
)
4198 yield from self
._nsrs
[nsr_id
].unpublish(xact
)
4200 # Finaly delete the NS instance from this NS Manager
4201 self
._log
.debug("Deletng the network service %s", nsr_id
)
4202 self
.delete_nsr(nsr_id
)
4205 class NsmRecordsPublisherProxy(object):
4206 """ This class provides a publisher interface that allows plugin objects
4207 to publish NSR/VNFR/VLR"""
4209 def __init__(self
, dts
, log
, loop
, nsr_pub_hdlr
, vnfr_pub_hdlr
, vlr_pub_hdlr
):
4213 self
._nsr
_pub
_hdlr
= nsr_pub_hdlr
4214 self
._vlr
_pub
_hdlr
= vlr_pub_hdlr
4215 self
._vnfr
_pub
_hdlr
= vnfr_pub_hdlr
4218 def publish_nsr(self
, xact
, nsr
):
4219 """ Publish an NSR """
4220 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4221 return (yield from self
._nsr
_pub
_hdlr
.update(xact
, path
, nsr
))
4224 def unpublish_nsr(self
, xact
, nsr
):
4225 """ Unpublish an NSR """
4226 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4227 return (yield from self
._nsr
_pub
_hdlr
.delete(xact
, path
))
4230 def publish_vnfr(self
, xact
, vnfr
):
4231 """ Publish an VNFR """
4232 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4233 return (yield from self
._vnfr
_pub
_hdlr
.update(xact
, path
, vnfr
))
4236 def unpublish_vnfr(self
, xact
, vnfr
):
4237 """ Unpublish a VNFR """
4238 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4239 return (yield from self
._vnfr
_pub
_hdlr
.delete(xact
, path
))
4242 def publish_vlr(self
, xact
, vlr
):
4243 """ Publish a VLR """
4244 path
= VirtualLinkRecord
.vlr_xpath(vlr
)
4245 return (yield from self
._vlr
_pub
_hdlr
.update(xact
, path
, vlr
))
4248 def unpublish_vlr(self
, xact
, vlr
):
4249 """ Unpublish a VLR """
4250 path
= VirtualLinkRecord
.vlr_xpath(vlr
)
4251 return (yield from self
._vlr
_pub
_hdlr
.delete(xact
, path
))
4254 class ScalingRpcHandler(mano_dts
.DtsHandler
):
4255 """ The Network service Monitor DTS handler """
4256 SCALE_IN_INPUT_XPATH
= "I,/nsr:exec-scale-in"
4257 SCALE_IN_OUTPUT_XPATH
= "O,/nsr:exec-scale-in"
4259 SCALE_OUT_INPUT_XPATH
= "I,/nsr:exec-scale-out"
4260 SCALE_OUT_OUTPUT_XPATH
= "O,/nsr:exec-scale-out"
4262 ACTION
= Enum('ACTION', 'SCALE_IN SCALE_OUT')
4264 def __init__(self
, log
, dts
, loop
, callback
=None):
4265 super().__init
__(log
, dts
, loop
)
4266 self
.callback
= callback
4267 self
.last_instance_id
= defaultdict(int)
4273 def on_scale_in_prepare(xact_info
, action
, ks_path
, msg
):
4274 assert action
== rwdts
.QueryAction
.RPC
4277 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleIn
.from_dict({
4278 "instance_id": msg
.instance_id
})
4280 xact_info
.respond_xpath(
4281 rwdts
.XactRspCode
.ACK
,
4282 self
.__class
__.SCALE_IN_OUTPUT_XPATH
,
4286 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_IN
)
4287 except Exception as e
:
4288 self
.log
.exception(e
)
4289 xact_info
.respond_xpath(
4290 rwdts
.XactRspCode
.NACK
,
4291 self
.__class
__.SCALE_IN_OUTPUT_XPATH
)
4294 def on_scale_out_prepare(xact_info
, action
, ks_path
, msg
):
4295 assert action
== rwdts
.QueryAction
.RPC
4298 scaling_group
= msg
.scaling_group_name_ref
4299 if not msg
.instance_id
:
4300 last_instance_id
= self
.last_instance_id
[scale_group
]
4301 msg
.instance_id
= last_instance_id
+ 1
4302 self
.last_instance_id
[scale_group
] += 1
4304 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleOut
.from_dict({
4305 "instance_id": msg
.instance_id
})
4307 xact_info
.respond_xpath(
4308 rwdts
.XactRspCode
.ACK
,
4309 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
,
4313 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_OUT
)
4314 except Exception as e
:
4315 self
.log
.exception(e
)
4316 xact_info
.respond_xpath(
4317 rwdts
.XactRspCode
.NACK
,
4318 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
)
4320 scale_in_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4321 on_prepare
=on_scale_in_prepare
)
4322 scale_out_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4323 on_prepare
=on_scale_out_prepare
)
4325 with self
.dts
.group_create() as group
:
4327 xpath
=self
.__class
__.SCALE_IN_INPUT_XPATH
,
4328 handler
=scale_in_hdl
,
4329 flags
=rwdts
.Flag
.PUBLISHER
)
4331 xpath
=self
.__class
__.SCALE_OUT_INPUT_XPATH
,
4332 handler
=scale_out_hdl
,
4333 flags
=rwdts
.Flag
.PUBLISHER
)
4336 class NsmTasklet(rift
.tasklets
.Tasklet
):
4338 The network service manager tasklet
4340 def __init__(self
, *args
, **kwargs
):
4341 super(NsmTasklet
, self
).__init
__(*args
, **kwargs
)
4342 self
.rwlog
.set_category("rw-mano-log")
4343 self
.rwlog
.set_subcategory("nsm")
4348 self
._ro
_plugin
_selector
= None
4349 self
._vnffgmgr
= None
4351 self
._nsr
_handler
= None
4352 self
._vnfr
_pub
_handler
= None
4353 self
._vlr
_pub
_handler
= None
4354 self
._vnfd
_pub
_handler
= None
4355 self
._scale
_cfg
_handler
= None
4357 self
._records
_publisher
_proxy
= None
4360 """ The task start callback """
4361 super(NsmTasklet
, self
).start()
4362 self
.log
.info("Starting NsmTasklet")
4364 self
.log
.debug("Registering with dts")
4365 self
._dts
= rift
.tasklets
.DTS(self
.tasklet_info
,
4366 RwNsmYang
.get_schema(),
4368 self
.on_dts_state_change
)
4370 self
.log
.debug("Created DTS Api GI Object: %s", self
._dts
)
4376 print("Caught Exception in NSM stop:", sys
.exc_info()[0])
4379 def on_instance_started(self
):
4380 """ Task instance started callback """
4381 self
.log
.debug("Got instance started callback")
4385 """ Task init callback """
4386 self
.log
.debug("Got instance started callback")
4388 self
.log
.debug("creating config account handler")
4390 self
._nsr
_pub
_handler
= publisher
.NsrOpDataDtsHandler(self
._dts
, self
.log
, self
.loop
)
4391 yield from self
._nsr
_pub
_handler
.register()
4393 self
._vnfr
_pub
_handler
= publisher
.VnfrPublisherDtsHandler(self
._dts
, self
.log
, self
.loop
)
4394 yield from self
._vnfr
_pub
_handler
.register()
4396 self
._vlr
_pub
_handler
= publisher
.VlrPublisherDtsHandler(self
._dts
, self
.log
, self
.loop
)
4397 yield from self
._vlr
_pub
_handler
.register()
4399 manifest
= self
.tasklet_info
.get_pb_manifest()
4400 use_ssl
= manifest
.bootstrap_phase
.rwsecurity
.use_ssl
4401 ssl_cert
= manifest
.bootstrap_phase
.rwsecurity
.cert
4402 ssl_key
= manifest
.bootstrap_phase
.rwsecurity
.key
4404 self
._vnfd
_pub
_handler
= publisher
.VnfdPublisher(use_ssl
, ssl_cert
, ssl_key
, self
.loop
)
4406 self
._records
_publisher
_proxy
= NsmRecordsPublisherProxy(
4410 self
._nsr
_pub
_handler
,
4411 self
._vnfr
_pub
_handler
,
4412 self
._vlr
_pub
_handler
,
4415 # Register the NSM to receive the nsm plugin
4416 # when cloud account is configured
4417 self
._ro
_plugin
_selector
= cloud
.ROAccountPluginSelector(
4421 self
._records
_publisher
_proxy
,
4423 yield from self
._ro
_plugin
_selector
.register()
4425 self
._cloud
_account
_handler
= cloud
.CloudAccountConfigSubscriber(
4430 yield from self
._cloud
_account
_handler
.register()
4432 self
._vnffgmgr
= rwvnffgmgr
.VnffgMgr(self
._dts
, self
.log
, self
.log_hdl
, self
.loop
)
4433 yield from self
._vnffgmgr
.register()
4435 self
._nsm
= NsManager(
4439 self
._nsr
_pub
_handler
,
4440 self
._vnfr
_pub
_handler
,
4441 self
._vlr
_pub
_handler
,
4442 self
._ro
_plugin
_selector
,
4444 self
._vnfd
_pub
_handler
,
4445 self
._cloud
_account
_handler
4448 yield from self
._nsm
.register()
4452 """ Task run callback """
4456 def on_dts_state_change(self
, state
):
4457 """Take action according to current dts state to transition
4458 application into the corresponding application state
4461 state - current dts state
4464 rwdts
.State
.INIT
: rwdts
.State
.REGN_COMPLETE
,
4465 rwdts
.State
.CONFIG
: rwdts
.State
.RUN
,
4469 rwdts
.State
.INIT
: self
.init
,
4470 rwdts
.State
.RUN
: self
.run
,
4473 # Transition application to next state
4474 handler
= handlers
.get(state
, None)
4475 if handler
is not None:
4476 yield from handler()
4478 # Transition dts to next state
4479 next_state
= switch
.get(state
, None)
4480 if next_state
is not None:
4481 self
.log
.debug("Changing state to %s", next_state
)
4482 self
._dts
.handle
.set_state(next_state
)