2 # Copyright 2016 RIFT.IO Inc
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
17 # vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
20 import ncclient
.asyncio_manager
32 from collections
import deque
33 from collections
import defaultdict
37 gi
.require_version('RwYang', '1.0')
38 gi
.require_version('RwNsdYang', '1.0')
39 gi
.require_version('RwDts', '1.0')
40 gi
.require_version('RwNsmYang', '1.0')
41 gi
.require_version('RwNsrYang', '1.0')
42 gi
.require_version('RwTypes', '1.0')
43 gi
.require_version('RwVlrYang', '1.0')
44 gi
.require_version('RwVnfrYang', '1.0')
45 from gi
.repository
import (
61 import rift
.mano
.ncclient
62 import rift
.mano
.config_data
.config
63 import rift
.mano
.dts
as mano_dts
65 from . import rwnsm_conman
as conman
67 from . import publisher
69 from . import config_value_pool
70 from . import rwvnffgmgr
71 from . import scale_group
74 class NetworkServiceRecordState(Enum
):
75 """ Network Service Record State """
79 VNFFG_INIT_PHASE
= 104
85 VL_TERMINATE_PHASE
= 111
86 VNF_TERMINATE_PHASE
= 112
87 VNFFG_TERMINATE_PHASE
= 113
94 class NetworkServiceRecordError(Exception):
95 """ Network Service Record Error """
99 class NetworkServiceDescriptorError(Exception):
100 """ Network Service Descriptor Error """
104 class VirtualNetworkFunctionRecordError(Exception):
105 """ Virtual Network Function Record Error """
109 class NetworkServiceDescriptorNotFound(Exception):
110 """ Cannot find Network Service Descriptor"""
114 class NetworkServiceDescriptorRefCountExists(Exception):
115 """ Network Service Descriptor reference count exists """
119 class NetworkServiceDescriptorUnrefError(Exception):
120 """ Failed to unref a network service descriptor """
124 class NsrInstantiationFailed(Exception):
125 """ Failed to instantiate network service """
129 class VnfInstantiationFailed(Exception):
130 """ Failed to instantiate virtual network function"""
134 class VnffgInstantiationFailed(Exception):
135 """ Failed to instantiate virtual network function"""
139 class VnfDescriptorError(Exception):
140 """Failed to instantiate virtual network function"""
144 class ScalingOperationError(Exception):
148 class ScaleGroupMissingError(Exception):
152 class PlacementGroupError(Exception):
156 class NsrNsdUpdateError(Exception):
160 class NsrVlUpdateError(NsrNsdUpdateError
):
164 class VlRecordState(Enum
):
165 """ VL Record State """
167 INSTANTIATION_PENDING
= 102
169 TERMINATE_PENDING
= 104
174 class VnffgRecordState(Enum
):
175 """ VNFFG Record State """
177 INSTANTIATION_PENDING
= 102
179 TERMINATE_PENDING
= 104
184 class VnffgRecord(object):
185 """ Vnffg Records class"""
188 def __init__(self
, dts
, log
, loop
, vnffgmgr
, nsr
, nsr_name
, vnffgd_msg
, sdn_account_name
):
193 self
._vnffgmgr
= vnffgmgr
195 self
._nsr
_name
= nsr_name
196 self
._vnffgd
_msg
= vnffgd_msg
197 if sdn_account_name
is None:
198 self
._sdn
_account
_name
= ''
200 self
._sdn
_account
_name
= sdn_account_name
202 self
._vnffgr
_id
= str(uuid
.uuid4())
203 self
._vnffgr
_rsp
_id
= list()
204 self
._vnffgr
_state
= VnffgRecordState
.INIT
209 return self
._vnffgr
_id
213 """ state of this VNF """
214 return self
._vnffgr
_state
216 def fetch_vnffgr(self
):
218 Get VNFFGR message to be published
221 if self
._vnffgr
_state
== VnffgRecordState
.INIT
:
222 vnffgr_dict
= {"id": self
._vnffgr
_id
,
223 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
224 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
225 "sdn_account": self
._sdn
_account
_name
,
226 "operational_status": 'init',
228 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
229 elif self
._vnffgr
_state
== VnffgRecordState
.TERMINATED
:
230 vnffgr_dict
= {"id": self
._vnffgr
_id
,
231 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
232 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
233 "sdn_account": self
._sdn
_account
_name
,
234 "operational_status": 'terminated',
236 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
239 vnffgr
= self
._vnffgmgr
.fetch_vnffgr(self
._vnffgr
_id
)
241 self
._log
.exception("Fetching VNFFGR for VNFFG with id %s failed", self
._vnffgr
_id
)
242 self
._vnffgr
_state
= VnffgRecordState
.FAILED
243 vnffgr_dict
= {"id": self
._vnffgr
_id
,
244 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
245 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
246 "sdn_account": self
._sdn
_account
_name
,
247 "operational_status": 'failed',
249 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
254 def vnffgr_create_msg(self
):
255 """ Virtual Link Record message for Creating VLR in VNS """
256 vnffgr_dict
= {"id": self
._vnffgr
_id
,
257 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
258 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
259 "sdn_account": self
._sdn
_account
_name
,
261 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
262 for rsp
in self
._vnffgd
_msg
.rsp
:
263 vnffgr_rsp
= vnffgr
.rsp
.add()
264 vnffgr_rsp
.id = str(uuid
.uuid4())
265 vnffgr_rsp
.name
= self
._nsr
.name
+ '.' + rsp
.name
266 self
._vnffgr
_rsp
_id
.append(vnffgr_rsp
.id)
267 vnffgr_rsp
.vnffgd_rsp_id_ref
= rsp
.id
268 vnffgr_rsp
.vnffgd_rsp_name_ref
= rsp
.name
269 for rsp_cp_ref
in rsp
.vnfd_connection_point_ref
:
270 vnfd
= [vnfr
.vnfd
for vnfr
in self
._nsr
.vnfrs
.values() if vnfr
.vnfd
.id == rsp_cp_ref
.vnfd_id_ref
]
271 self
._log
.debug("VNFD message during VNFFG instantiation is %s",vnfd
)
272 if len(vnfd
) > 0 and vnfd
[0].has_field('service_function_type'):
273 self
._log
.debug("Service Function Type for VNFD ID %s is %s",rsp_cp_ref
.vnfd_id_ref
, vnfd
[0].service_function_type
)
275 self
._log
.error("Service Function Type not available for VNFD ID %s; Skipping in chain",rsp_cp_ref
.vnfd_id_ref
)
278 vnfr_cp_ref
= vnffgr_rsp
.vnfr_connection_point_ref
.add()
279 vnfr_cp_ref
.member_vnf_index_ref
= rsp_cp_ref
.member_vnf_index_ref
280 vnfr_cp_ref
.hop_number
= rsp_cp_ref
.order
281 vnfr_cp_ref
.vnfd_id_ref
=rsp_cp_ref
.vnfd_id_ref
282 vnfr_cp_ref
.service_function_type
= vnfd
[0].service_function_type
283 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
284 if (nsr_vnfr
.vnfd
.id == vnfr_cp_ref
.vnfd_id_ref
and
285 nsr_vnfr
.member_vnf_index
== vnfr_cp_ref
.member_vnf_index_ref
):
286 vnfr_cp_ref
.vnfr_id_ref
= nsr_vnfr
.id
287 vnfr_cp_ref
.vnfr_name_ref
= nsr_vnfr
.name
288 vnfr_cp_ref
.vnfr_connection_point_ref
= rsp_cp_ref
.vnfd_connection_point_ref
290 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
291 self
._log
.debug(" Received VNFR is %s", vnfr
)
292 while vnfr
.operational_status
!= 'running':
293 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
294 if vnfr
.operational_status
== 'failed':
295 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
296 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self
.id, vnfr
.id))
297 yield from asyncio
.sleep(2, loop
=self
._loop
)
298 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
299 self
._log
.debug("Received VNFR is %s", vnfr
)
301 vnfr_cp_ref
.connection_point_params
.mgmt_address
= vnfr
.mgmt_interface
.ip_address
302 for cp
in vnfr
.connection_point
:
303 if cp
.name
== vnfr_cp_ref
.vnfr_connection_point_ref
:
304 vnfr_cp_ref
.connection_point_params
.port_id
= cp
.connection_point_id
305 vnfr_cp_ref
.connection_point_params
.name
= self
._nsr
.name
+ '.' + cp
.name
306 for vdu
in vnfr
.vdur
:
307 for ext_intf
in vdu
.external_interface
:
308 if ext_intf
.name
== vnfr_cp_ref
.vnfr_connection_point_ref
:
309 vnfr_cp_ref
.connection_point_params
.vm_id
= vdu
.vim_id
310 self
._log
.debug("VIM ID for CP %s in VNFR %s is %s",cp
.name
,nsr_vnfr
.id,
311 vnfr_cp_ref
.connection_point_params
.vm_id
)
314 vnfr_cp_ref
.connection_point_params
.address
= cp
.ip_address
315 vnfr_cp_ref
.connection_point_params
.port
= VnffgRecord
.SFF_DP_PORT
317 for vnffgd_classifier
in self
._vnffgd
_msg
.classifier
:
318 _rsp
= [rsp
for rsp
in vnffgr
.rsp
if rsp
.vnffgd_rsp_id_ref
== vnffgd_classifier
.rsp_id_ref
]
320 rsp_id_ref
= _rsp
[0].id
321 rsp_name
= _rsp
[0].name
323 self
._log
.error("RSP with ID %s not found during classifier creation for classifier id %s",vnffgd_classifier
.rsp_id_ref
,vnffgd_classifier
.id)
325 vnffgr_classifier
= vnffgr
.classifier
.add()
326 vnffgr_classifier
.id = vnffgd_classifier
.id
327 vnffgr_classifier
.name
= self
._nsr
.name
+ '.' + vnffgd_classifier
.name
328 _rsp
[0].classifier_name
= vnffgr_classifier
.name
329 vnffgr_classifier
.rsp_id_ref
= rsp_id_ref
330 vnffgr_classifier
.rsp_name
= rsp_name
331 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
332 if (nsr_vnfr
.vnfd
.id == vnffgd_classifier
.vnfd_id_ref
and
333 nsr_vnfr
.member_vnf_index
== vnffgd_classifier
.member_vnf_index_ref
):
334 vnffgr_classifier
.vnfr_id_ref
= nsr_vnfr
.id
335 vnffgr_classifier
.vnfr_name_ref
= nsr_vnfr
.name
336 vnffgr_classifier
.vnfr_connection_point_ref
= vnffgd_classifier
.vnfd_connection_point_ref
338 if nsr_vnfr
.vnfd
.service_function_chain
== 'CLASSIFIER':
339 vnffgr_classifier
.sff_name
= nsr_vnfr
.name
341 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
342 self
._log
.debug(" Received VNFR is %s", vnfr
)
343 while vnfr
.operational_status
!= 'running':
344 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
345 if vnfr
.operational_status
== 'failed':
346 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
347 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self
.id, vnfr
.id))
348 yield from asyncio
.sleep(2, loop
=self
._loop
)
349 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
350 self
._log
.debug("Received VNFR is %s", vnfr
)
352 for cp
in vnfr
.connection_point
:
353 if cp
.name
== vnffgr_classifier
.vnfr_connection_point_ref
:
354 vnffgr_classifier
.port_id
= cp
.connection_point_id
355 vnffgr_classifier
.ip_address
= cp
.ip_address
356 for vdu
in vnfr
.vdur
:
357 for ext_intf
in vdu
.external_interface
:
358 if ext_intf
.name
== vnffgr_classifier
.vnfr_connection_point_ref
:
359 vnffgr_classifier
.vm_id
= vdu
.vim_id
360 self
._log
.debug("VIM ID for CP %s in VNFR %s is %s",cp
.name
,nsr_vnfr
.id,
361 vnfr_cp_ref
.connection_point_params
.vm_id
)
364 self
._log
.info("VNFFGR msg to be sent is %s", vnffgr
)
368 def vnffgr_nsr_sff_list(self
):
369 """ SFF List for VNFR """
371 sf_list
= [nsr_vnfr
.name
for nsr_vnfr
in self
._nsr
.vnfrs
.values() if nsr_vnfr
.vnfd
.service_function_chain
== 'SF']
373 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
374 if (nsr_vnfr
.vnfd
.service_function_chain
== 'CLASSIFIER' or nsr_vnfr
.vnfd
.service_function_chain
== 'SFF'):
375 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
376 self
._log
.debug(" Received VNFR is %s", vnfr
)
377 while vnfr
.operational_status
!= 'running':
378 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
379 if vnfr
.operational_status
== 'failed':
380 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
381 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self
.id, vnfr
.id))
382 yield from asyncio
.sleep(2, loop
=self
._loop
)
383 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
384 self
._log
.debug("Received VNFR is %s", vnfr
)
386 sff
= RwsdnYang
.VNFFGSff()
387 sff_list
[nsr_vnfr
.vnfd
.id] = sff
388 sff
.name
= nsr_vnfr
.name
389 sff
.function_type
= nsr_vnfr
.vnfd
.service_function_chain
391 sff
.mgmt_address
= vnfr
.mgmt_interface
.ip_address
392 sff
.mgmt_port
= VnffgRecord
.SFF_MGMT_PORT
393 for cp
in vnfr
.connection_point
:
394 sff_dp
= sff
.dp_endpoints
.add()
395 sff_dp
.name
= self
._nsr
.name
+ '.' + cp
.name
396 sff_dp
.address
= cp
.ip_address
397 sff_dp
.port
= VnffgRecord
.SFF_DP_PORT
398 if nsr_vnfr
.vnfd
.service_function_chain
== 'SFF':
399 for sf_name
in sf_list
:
400 _sf
= sff
.vnfr_list
.add()
401 _sf
.vnfr_name
= sf_name
406 def instantiate(self
):
407 """ Instantiate this VNFFG """
409 self
._log
.info("Instaniating VNFFGR with vnffgd %s",
413 vnffgr_request
= yield from self
.vnffgr_create_msg()
414 vnffg_sff_list
= yield from self
.vnffgr_nsr_sff_list()
417 vnffgr
= self
._vnffgmgr
.create_vnffgr(vnffgr_request
,self
._vnffgd
_msg
.classifier
,vnffg_sff_list
)
418 except Exception as e
:
419 self
._log
.exception("VNFFG instantiation failed: %s", str(e
))
420 self
._vnffgr
_state
= VnffgRecordState
.FAILED
421 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFFGR %s failure" % (self
.id, vnffgr_request
.id))
423 self
._vnffgr
_state
= VnffgRecordState
.INSTANTIATION_PENDING
425 self
._log
.info("Instantiated VNFFGR :%s", vnffgr
)
426 self
._vnffgr
_state
= VnffgRecordState
.ACTIVE
428 self
._log
.info("Invoking update_state to update NSR state for NSR ID: %s", self
._nsr
.id)
429 yield from self
._nsr
.update_state()
431 def vnffgr_in_vnffgrm(self
):
432 """ Is there a VNFR record in VNFM """
433 if (self
._vnffgr
_state
== VnffgRecordState
.ACTIVE
or
434 self
._vnffgr
_state
== VnffgRecordState
.INSTANTIATION_PENDING
or
435 self
._vnffgr
_state
== VnffgRecordState
.FAILED
):
442 """ Terminate this VNFFGR """
443 if not self
.vnffgr_in_vnffgrm():
444 self
._log
.error("Ignoring terminate request for id %s in state %s",
445 self
.id, self
._vnffgr
_state
)
448 self
._log
.info("Terminating VNFFGR id:%s", self
.id)
449 self
._vnffgr
_state
= VnffgRecordState
.TERMINATE_PENDING
451 self
._vnffgmgr
.terminate_vnffgr(self
._vnffgr
_id
)
453 self
._vnffgr
_state
= VnffgRecordState
.TERMINATED
454 self
._log
.debug("Terminated VNFFGR id:%s", self
.id)
457 class VirtualLinkRecord(object):
458 """ Virtual Link Records class"""
459 XPATH
= "D,/vlr:vlr-catalog/vlr:vlr"
462 def create_record(dts
, log
, loop
, nsr_name
, vld_msg
, cloud_account_name
, om_datacenter
, ip_profile
, nsr_id
, restart_mode
=False):
463 """Creates a new VLR object based on the given data.
465 If restart mode is enabled, then we look for existing records in the
466 DTS and create a VLR records using the exiting data(ID)
471 vlr_obj
= VirtualLinkRecord(
484 res_iter
= yield from dts
.query_read(
485 "D,/vlr:vlr-catalog/vlr:vlr",
486 rwdts
.XactFlag
.MERGE
)
489 response
= yield from fut
490 vlr
= response
.result
492 # Check if the record is already present, if so use the ID of
493 # the existing record. Since the name of the record is uniquely
494 # formed we can use it as a search key!
495 if vlr
.name
== vlr_obj
.name
:
496 vlr_obj
.reset_id(vlr
.id)
501 def __init__(self
, dts
, log
, loop
, nsr_name
, vld_msg
, cloud_account_name
, om_datacenter
, ip_profile
, nsr_id
):
505 self
._nsr
_name
= nsr_name
506 self
._vld
_msg
= vld_msg
507 self
._cloud
_account
_name
= cloud_account_name
508 self
._om
_datacenter
_name
= om_datacenter
509 self
._assigned
_subnet
= None
510 self
._nsr
_id
= nsr_id
511 self
._ip
_profile
= ip_profile
512 self
._vlr
_id
= str(uuid
.uuid4())
513 self
._state
= VlRecordState
.INIT
514 self
._prev
_state
= None
515 self
._create
_time
= int(time
.time())
519 """ path for this object """
520 return "D,/vlr:vlr-catalog/vlr:vlr[vlr:id = '{}']".format(self
._vlr
_id
)
529 """ Get NSR name for this VL """
534 """ Virtual Link Desciptor """
538 def assigned_subnet(self
):
539 """ Subnet assigned to this VL"""
540 return self
._assigned
_subnet
545 Get the name for this VLR.
546 VLR name is "nsr name:VLD name"
548 if self
.vld_msg
.vim_network_name
:
549 return self
.vld_msg
.vim_network_name
550 elif self
.vld_msg
.name
== "multisite":
551 # This is a temporary hack to identify manually provisioned inter-site network
552 return self
.vld_msg
.name
554 return self
._nsr
_name
+ "." + self
.vld_msg
.name
557 def cloud_account_name(self
):
558 """ Cloud account that this VLR should be created in """
559 return self
._cloud
_account
_name
562 def om_datacenter_name(self
):
563 """ Datacenter that this VLR should be created in """
564 return self
._om
_datacenter
_name
568 """ Get the VLR path from VLR """
569 return (VirtualLinkRecord
.XPATH
+ "[vlr:id = '{}']").format(vlr
.id)
577 def state(self
, value
):
578 """ VLR set state """
582 def prev_state(self
):
583 """ VLR previous state """
584 return self
._prev
_state
587 def prev_state(self
, value
):
588 """ VLR set previous state """
589 self
._prev
_state
= value
593 """ Virtual Link Record message for Creating VLR in VNS """
594 vld_fields
= ["short_name",
602 vld_copy_dict
= {k
: v
for k
, v
in self
.vld_msg
.as_dict().items()
605 vlr_dict
= {"id": self
._vlr
_id
,
606 "nsr_id_ref": self
._nsr
_id
,
607 "vld_ref": self
.vld_msg
.id,
609 "create_time": self
._create
_time
,
610 "cloud_account": self
.cloud_account_name
,
611 "om_datacenter": self
.om_datacenter_name
,
614 if self
._ip
_profile
and self
._ip
_profile
.has_field('ip_profile_params'):
615 vlr_dict
['ip_profile_params' ] = self
._ip
_profile
.ip_profile_params
.as_dict()
617 vlr_dict
.update(vld_copy_dict
)
618 vlr
= RwVlrYang
.YangData_Vlr_VlrCatalog_Vlr
.from_dict(vlr_dict
)
621 def reset_id(self
, vlr_id
):
622 self
._vlr
_id
= vlr_id
624 def create_nsr_vlr_msg(self
, vnfrs
):
625 """ The VLR message"""
626 nsr_vlr
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vlr()
627 nsr_vlr
.vlr_ref
= self
._vlr
_id
628 nsr_vlr
.assigned_subnet
= self
.assigned_subnet
629 nsr_vlr
.cloud_account
= self
.cloud_account_name
630 nsr_vlr
.om_datacenter
= self
.om_datacenter_name
632 for conn
in self
.vld_msg
.vnfd_connection_point_ref
:
634 if (vnfr
.vnfd
.id == conn
.vnfd_id_ref
and
635 vnfr
.member_vnf_index
== conn
.member_vnf_index_ref
and
636 self
.cloud_account_name
== vnfr
.cloud_account_name
and
637 self
.om_datacenter_name
== vnfr
.om_datacenter_name
):
638 cp_entry
= nsr_vlr
.vnfr_connection_point_ref
.add()
639 cp_entry
.vnfr_id
= vnfr
.id
640 cp_entry
.connection_point
= conn
.vnfd_connection_point_ref
645 def instantiate(self
):
646 """ Instantiate this VL """
647 self
._log
.debug("Instaniating VLR key %s, vld %s",
648 self
.xpath
, self
._vld
_msg
)
650 self
._state
= VlRecordState
.INSTANTIATION_PENDING
651 self
._log
.debug("Executing VL create path:%s msg:%s",
652 self
.xpath
, self
.vlr_msg
)
654 with self
._dts
.transaction(flags
=0) as xact
:
655 block
= xact
.block_create()
656 block
.add_query_create(self
.xpath
, self
.vlr_msg
)
657 self
._log
.debug("Executing VL create path:%s msg:%s",
658 self
.xpath
, self
.vlr_msg
)
659 res_iter
= yield from block
.execute(now
=True)
665 self
._state
= VlRecordState
.FAILED
666 raise NsrInstantiationFailed("Failed NS %s instantiation due to empty response" % self
.id)
668 if vlr
.operational_status
== 'failed':
669 self
._log
.debug("NS Id:%s VL creation failed for vlr id %s", self
.id, vlr
.id)
670 self
._state
= VlRecordState
.FAILED
671 raise NsrInstantiationFailed("Failed VL %s instantiation (%s)" % (vlr
.id, vlr
.operational_status_details
))
673 self
._log
.info("Instantiated VL with xpath %s and vlr:%s",
675 self
._state
= VlRecordState
.ACTIVE
676 self
._assigned
_subnet
= vlr
.assigned_subnet
678 def vlr_in_vns(self
):
679 """ Is there a VLR record in VNS """
680 if (self
._state
== VlRecordState
.ACTIVE
or
681 self
._state
== VlRecordState
.INSTANTIATION_PENDING
or
682 self
._state
== VlRecordState
.TERMINATE_PENDING
or
683 self
._state
== VlRecordState
.FAILED
):
690 """ Terminate this VL """
691 if not self
.vlr_in_vns():
692 self
._log
.debug("Ignoring terminate request for id %s in state %s",
693 self
.id, self
._state
)
696 self
._log
.debug("Terminating VL id:%s", self
.id)
697 self
._state
= VlRecordState
.TERMINATE_PENDING
699 with self
._dts
.transaction(flags
=0) as xact
:
700 block
= xact
.block_create()
701 block
.add_query_delete(self
.xpath
)
702 yield from block
.execute(flags
=0, now
=True)
704 self
._state
= VlRecordState
.TERMINATED
705 self
._log
.debug("Terminated VL id:%s", self
.id)
708 class VnfRecordState(Enum
):
709 """ Vnf Record State """
711 INSTANTIATION_PENDING
= 102
713 TERMINATE_PENDING
= 104
718 class VirtualNetworkFunctionRecord(object):
719 """ Virtual Network Function Record class"""
720 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
724 def create_record(dts
, log
, loop
, vnfd
, const_vnfd_msg
, nsd_id
, nsr_name
,
725 cloud_account_name
, om_datacenter_name
, nsr_id
, group_name
, group_instance_id
,
726 placement_groups
, restart_mode
=False):
727 """Creates a new VNFR object based on the given data.
729 If restart mode is enabled, then we look for existing records in the
730 DTS and create a VNFR records using the exiting data(ID)
733 VirtualNetworkFunctionRecord
735 vnfr_obj
= VirtualNetworkFunctionRecord(
749 restart_mode
=restart_mode
)
752 res_iter
= yield from dts
.query_read(
753 "D,/vnfr:vnfr-catalog/vnfr:vnfr",
754 rwdts
.XactFlag
.MERGE
)
757 response
= yield from fut
758 vnfr
= response
.result
760 if vnfr
.name
== vnfr_obj
.name
:
761 vnfr_obj
.reset_id(vnfr
.id)
778 group_instance_id
=None,
779 placement_groups
= [],
780 restart_mode
= False):
785 self
._const
_vnfd
_msg
= const_vnfd_msg
786 self
._nsd
_id
= nsd_id
787 self
._nsr
_name
= nsr_name
788 self
._nsr
_id
= nsr_id
789 self
._cloud
_account
_name
= cloud_account_name
790 self
._om
_datacenter
_name
= om_datacenter_name
791 self
._group
_name
= group_name
792 self
._group
_instance
_id
= group_instance_id
793 self
._placement
_groups
= placement_groups
794 self
._config
_status
= NsrYang
.ConfigStates
.INIT
795 self
._create
_time
= int(time
.time())
797 self
._prev
_state
= VnfRecordState
.INIT
798 self
._state
= VnfRecordState
.INIT
799 self
._state
_failed
_reason
= None
801 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
804 self
._vnfr
_id
= str(uuid
.uuid4())
806 self
._vnfr
_msg
= self
.create_vnfr_msg()
807 self
._log
.debug("Set VNFR {} config type to {}".
808 format(self
.name
, self
.config_type
))
809 self
.restart_mode
= restart_mode
812 if group_name
is None and group_instance_id
is not None:
813 raise ValueError("Group instance id must not be provided with an empty group name")
823 return "D,/vnfr:vnfr-catalog/vnfr:vnfr[vnfr:id = '{}']".format(self
.id)
828 return self
._vnfr
_msg
831 def const_vnfr_msg(self
):
833 return RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_ConstituentVnfrRef(vnfr_id
=self
.id,cloud_account
=self
.cloud_account_name
,om_datacenter
=self
._om
_datacenter
_name
)
841 def cloud_account_name(self
):
842 """ Cloud account that this VNF should be created in """
843 return self
._cloud
_account
_name
846 def om_datacenter_name(self
):
847 """ Datacenter that this VNF should be created in """
848 return self
._om
_datacenter
_name
853 """ Is this VNF actve """
854 return True if self
._state
== VnfRecordState
.ACTIVE
else False
858 """ state of this VNF """
862 def state_failed_reason(self
):
863 """ Error message in case this VNF is in failed state """
864 return self
._state
_failed
_reason
867 def member_vnf_index(self
):
868 """ Member VNF index """
869 return self
._const
_vnfd
_msg
.member_vnf_index
874 return self
._nsr
_name
878 """ Name of this VNFR """
879 if self
._name
is not None:
882 name_tags
= [self
._nsr
_name
]
884 if self
._group
_name
is not None:
885 name_tags
.append(self
._group
_name
)
887 if self
._group
_instance
_id
is not None:
888 name_tags
.append(str(self
._group
_instance
_id
))
890 name_tags
.extend([self
.vnfd
.name
, str(self
.member_vnf_index
)])
892 self
._name
= "__".join(name_tags
)
897 def vnfr_xpath(vnfr
):
898 """ Get the VNFR path from VNFR """
899 return (VirtualNetworkFunctionRecord
.XPATH
+ "[vnfr:id = '{}']").format(vnfr
.id)
902 def config_type(self
):
903 cfg_types
= ['netconf', 'juju', 'script']
904 for method
in cfg_types
:
905 if self
._vnfd
.vnf_configuration
.has_field(method
):
910 def config_status(self
):
911 """Return the config status as YANG ENUM string"""
912 self
._log
.debug("Map VNFR {} config status {} ({})".
913 format(self
.name
, self
._config
_status
, self
.config_type
))
914 if self
.config_type
== 'none':
915 return 'config_not_needed'
916 elif self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
918 elif self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
923 def set_state(self
, state
):
924 """ set the state of this object """
925 self
._prev
_state
= self
._state
928 def reset_id(self
, vnfr_id
):
929 self
._vnfr
_id
= vnfr_id
930 self
._vnfr
_msg
= self
.create_vnfr_msg()
933 self
.config_store
.merge_vnfd_config(
936 self
.member_vnf_index
,
939 def create_vnfr_msg(self
):
940 """ VNFR message for this VNFR """
948 vnfd_copy_dict
= {k
: v
for k
, v
in self
._vnfd
.as_dict().items() if k
in vnfd_fields
}
951 "nsr_id_ref": self
._nsr
_id
,
953 "cloud_account": self
._cloud
_account
_name
,
954 "om_datacenter": self
._om
_datacenter
_name
,
955 "config_status": self
.config_status
957 vnfr_dict
.update(vnfd_copy_dict
)
959 vnfr
= RwVnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr
.from_dict(vnfr_dict
)
961 vnfr
.vnfd
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_Vnfd
.from_dict(self
.vnfd
.as_dict(),
962 ignore_missing_keys
=True)
963 vnfr
.member_vnf_index_ref
= self
.member_vnf_index
964 vnfr
.vnf_configuration
.from_dict(self
._vnfd
.vnf_configuration
.as_dict())
966 if self
._vnfd
.mgmt_interface
.has_field("port"):
967 vnfr
.mgmt_interface
.port
= self
._vnfd
.mgmt_interface
.port
969 for group_info
in self
._placement
_groups
:
970 group
= vnfr
.placement_groups_info
.add()
971 group
.from_dict(group_info
.as_dict())
973 # UI expects the monitoring param field to exist
974 vnfr
.monitoring_param
= []
976 self
._log
.debug("Get vnfr_msg for VNFR {} : {}".format(self
.name
, vnfr
))
980 def update_vnfm(self
):
981 self
._log
.debug("Send an update to VNFM for VNFR {} with {}".
982 format(self
.name
, self
.vnfr_msg
))
983 yield from self
._dts
.query_update(
985 rwdts
.XactFlag
.TRACE
,
989 def get_config_status(self
):
990 """Return the config status as YANG ENUM"""
991 return self
._config
_status
994 def set_config_status(self
, status
):
996 def status_to_string(status
):
998 NsrYang
.ConfigStates
.INIT
: 'init',
999 NsrYang
.ConfigStates
.CONFIGURING
: 'configuring',
1000 NsrYang
.ConfigStates
.CONFIG_NOT_NEEDED
: 'config_not_needed',
1001 NsrYang
.ConfigStates
.CONFIGURED
: 'configured',
1002 NsrYang
.ConfigStates
.FAILED
: 'failed',
1005 return status_dc
[status
]
1007 self
._log
.debug("Update VNFR {} from {} ({}) to {}".
1008 format(self
.name
, self
._config
_status
,
1009 self
.config_type
, status
))
1010 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1011 self
._log
.error("Updating already configured VNFR {}".
1015 if self
._config
_status
!= status
:
1017 self
._config
_status
= status
1018 # I don't think this is used. Original implementor can check.
1019 # Caused Exception, so corrected it by status_to_string
1020 # But not sure whats the use of this variable?
1021 self
.vnfr_msg
.config_status
= status_to_string(status
)
1022 except Exception as e
:
1023 self
._log
.error("Exception=%s", str(e
))
1026 self
._log
.debug("Updated VNFR {} status to {}".format(self
.name
, status
))
1028 if self
._config
_status
!= NsrYang
.ConfigStates
.INIT
:
1030 # Publish only after VNFM has the VNFR created
1031 yield from self
.update_vnfm()
1032 except Exception as e
:
1033 self
._log
.error("Exception updating VNFM with new status {} of VNFR {}: {}".
1034 format(status
, self
.name
, e
))
1035 self
._log
.exception(e
)
1037 def is_configured(self
):
1038 if self
.config_type
== 'none':
1041 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1047 def instantiate(self
, nsr
):
1048 """ Instantiate this VNFR"""
1050 self
._log
.debug("Instaniating VNFR key %s, vnfd %s",
1051 self
.xpath
, self
._vnfd
)
1053 self
._log
.debug("Create VNF with xpath %s and vnfr %s",
1054 self
.xpath
, self
.vnfr_msg
)
1056 self
.set_state(VnfRecordState
.INSTANTIATION_PENDING
)
1058 def find_vlr_for_cp(conn
):
1059 """ Find VLR for the given connection point """
1060 for vlr
in nsr
.vlrs
:
1061 for vnfd_cp
in vlr
.vld_msg
.vnfd_connection_point_ref
:
1062 if (vnfd_cp
.vnfd_id_ref
== self
._vnfd
.id and
1063 vnfd_cp
.vnfd_connection_point_ref
== conn
.name
and
1064 vnfd_cp
.member_vnf_index_ref
== self
.member_vnf_index
and
1065 vlr
.cloud_account_name
== self
.cloud_account_name
):
1066 self
._log
.debug("Found VLR for cp_name:%s and vnf-index:%d",
1067 conn
.name
, self
.member_vnf_index
)
1071 # For every connection point in the VNFD fill in the identifier
1072 for conn_p
in self
._vnfd
.connection_point
:
1073 cpr
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_ConnectionPoint()
1074 cpr
.name
= conn_p
.name
1075 cpr
.type_yang
= conn_p
.type_yang
1076 cpr
.port_security_enabled
= conn_p
.port_security_enabled
1077 vlr_ref
= find_vlr_for_cp(conn_p
)
1079 msg
= "Failed to find VLR for cp = %s" % conn_p
.name
1080 self
._log
.debug("%s", msg
)
1081 # raise VirtualNetworkFunctionRecordError(msg)
1084 cpr
.vlr_ref
= vlr_ref
.id
1085 self
.vnfr_msg
.connection_point
.append(cpr
)
1086 self
._log
.debug("Connection point [%s] added, vnf id=%s vnfd id=%s",
1087 cpr
, self
.vnfr_msg
.id, self
.vnfr_msg
.vnfd
.id)
1089 if not self
.restart_mode
:
1090 yield from self
._dts
.query_create(self
.xpath
,
1094 yield from self
._dts
.query_update(self
.xpath
,
1098 self
._log
.info("Created VNF with xpath %s and vnfr %s",
1099 self
.xpath
, self
.vnfr_msg
)
1101 self
._log
.info("Instantiated VNFR with xpath %s and vnfd %s, vnfr %s",
1102 self
.xpath
, self
._vnfd
, self
.vnfr_msg
)
1105 def update_state(self
, vnfr_msg
):
1106 """ Update this VNFR"""
1107 if vnfr_msg
.operational_status
== "running":
1108 if self
.vnfr_msg
.operational_status
!= "running":
1109 yield from self
.is_active()
1110 elif vnfr_msg
.operational_status
== "failed":
1111 yield from self
.instantiation_failed(failed_reason
=vnfr_msg
.operational_status_details
)
1114 def is_active(self
):
1115 """ This VNFR is active """
1116 self
._log
.debug("VNFR %s is active", self
._vnfr
_id
)
1117 self
.set_state(VnfRecordState
.ACTIVE
)
1120 def instantiation_failed(self
, failed_reason
=None):
1121 """ This VNFR instantiation failed"""
1122 self
._log
.error("VNFR %s instantiation failed", self
._vnfr
_id
)
1123 self
.set_state(VnfRecordState
.FAILED
)
1124 self
._state
_failed
_reason
= failed_reason
1126 def vnfr_in_vnfm(self
):
1127 """ Is there a VNFR record in VNFM """
1128 if (self
._state
== VnfRecordState
.ACTIVE
or
1129 self
._state
== VnfRecordState
.INSTANTIATION_PENDING
or
1130 self
._state
== VnfRecordState
.FAILED
):
1136 def terminate(self
):
1137 """ Terminate this VNF """
1138 if not self
.vnfr_in_vnfm():
1139 self
._log
.debug("Ignoring terminate request for id %s in state %s",
1140 self
.id, self
._state
)
1143 self
._log
.debug("Terminating VNF id:%s", self
.id)
1144 self
.set_state(VnfRecordState
.TERMINATE_PENDING
)
1145 with self
._dts
.transaction(flags
=0) as xact
:
1146 block
= xact
.block_create()
1147 block
.add_query_delete(self
.xpath
)
1148 yield from block
.execute(flags
=0)
1149 self
.set_state(VnfRecordState
.TERMINATED
)
1150 self
._log
.debug("Terminated VNF id:%s", self
.id)
1153 class NetworkServiceStatus(object):
1154 """ A class representing the Network service's status """
1155 MAX_EVENTS_RECORDED
= 10
1156 """ Network service Status class"""
1157 def __init__(self
, dts
, log
, loop
):
1162 self
._state
= NetworkServiceRecordState
.INIT
1163 self
._events
= deque([])
1166 def create_notification(self
, evt
, evt_desc
, evt_details
):
1167 xp
= "N,/rw-nsr:nsm-notification"
1168 notif
= RwNsrYang
.YangNotif_RwNsr_NsmNotification()
1170 notif
.description
= evt_desc
1171 notif
.details
= evt_details
if evt_details
is not None else None
1173 yield from self
._dts
.query_create(xp
, rwdts
.XactFlag
.ADVISE
, notif
)
1174 self
._log
.info("Notification called by creating dts query: %s", notif
)
1176 def record_event(self
, evt
, evt_desc
, evt_details
):
1177 """ Record an event """
1178 self
._log
.debug("Recording event - evt %s, evt_descr %s len = %s",
1179 evt
, evt_desc
, len(self
._events
))
1180 if len(self
._events
) >= NetworkServiceStatus
.MAX_EVENTS_RECORDED
:
1181 self
._events
.popleft()
1182 self
._events
.append((int(time
.time()), evt
, evt_desc
,
1183 evt_details
if evt_details
is not None else None))
1185 self
._loop
.create_task(self
.create_notification(evt
,evt_desc
,evt_details
))
1187 def set_state(self
, state
):
1188 """ set the state of this status object """
1192 """ Return the state as a yang enum string """
1193 state_to_str_map
= {"INIT": "init",
1194 "VL_INIT_PHASE": "vl_init_phase",
1195 "VNF_INIT_PHASE": "vnf_init_phase",
1196 "VNFFG_INIT_PHASE": "vnffg_init_phase",
1197 "SCALING_GROUP_INIT_PHASE": "scaling_group_init_phase",
1198 "RUNNING": "running",
1199 "SCALING_OUT": "scaling_out",
1200 "SCALING_IN": "scaling_in",
1201 "TERMINATE_RCVD": "terminate_rcvd",
1202 "TERMINATE": "terminate",
1203 "VL_TERMINATE_PHASE": "vl_terminate_phase",
1204 "VNF_TERMINATE_PHASE": "vnf_terminate_phase",
1205 "VNFFG_TERMINATE_PHASE": "vnffg_terminate_phase",
1206 "TERMINATED": "terminated",
1208 "VL_INSTANTIATE": "vl_instantiate",
1209 "VL_TERMINATE": "vl_terminate",
1211 return state_to_str_map
[self
._state
.name
]
1215 """ State of this status object """
1220 """ Network Service Record as a message"""
1223 for entry
in self
._events
:
1224 event
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_OperationalEvents()
1227 event
.timestamp
, event
.event
, event
.description
, event
.details
= entry
1228 event_list
.append(event
)
1232 class NetworkServiceRecord(object):
1233 """ Network service record """
1234 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
1236 def __init__(self
, dts
, log
, loop
, nsm
, nsm_plugin
, nsr_cfg_msg
, sdn_account_name
, key_pairs
, restart_mode
=False,
1242 self
._nsr
_cfg
_msg
= nsr_cfg_msg
1243 self
._nsm
_plugin
= nsm_plugin
1244 self
._sdn
_account
_name
= sdn_account_name
1245 self
._vlr
_handler
= vlr_handler
1248 self
._nsr
_msg
= None
1249 self
._nsr
_regh
= None
1250 self
._key
_pairs
= key_pairs
1255 self
._param
_pools
= {}
1256 self
._scaling
_groups
= {}
1257 self
._create
_time
= int(time
.time())
1258 self
._op
_status
= NetworkServiceStatus(dts
, log
, loop
)
1259 self
._config
_status
= NsrYang
.ConfigStates
.CONFIGURING
1260 self
._config
_status
_details
= None
1262 self
.restart_mode
= restart_mode
1263 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
1264 self
._debug
_running
= False
1265 self
._is
_active
= False
1266 self
._vl
_phase
_completed
= False
1267 self
._vnf
_phase
_completed
= False
1268 self
.vlr_uptime_tasks
= {}
1271 # Initalise the state to init
1272 # The NSR moves through the following transitions
1273 # 1. INIT -> VLS_READY once all the VLs in the NSD are created
1274 # 2. VLS_READY - VNFS_READY when all the VNFs in the NSD are created
1275 # 3. VNFS_READY - READY when the NSR is published
1277 self
.set_state(NetworkServiceRecordState
.INIT
)
1279 self
.substitute_input_parameters
= InputParameterSubstitution(self
._log
)
1282 def nsm_plugin(self
):
1284 return self
._nsm
_plugin
1286 def set_state(self
, state
):
1287 """ Set state for this NSR"""
1288 self
._log
.debug("Setting state to %s", state
)
1289 # We are in init phase and is moving to the next state
1290 # The new state could be a FAILED state or VNF_INIIT_PHASE
1291 if self
.state
== NetworkServiceRecordState
.VL_INIT_PHASE
:
1292 self
._vl
_phase
_completed
= True
1294 if self
.state
== NetworkServiceRecordState
.VNF_INIT_PHASE
:
1295 self
._vnf
_phase
_completed
= True
1297 self
._op
_status
.set_state(state
)
1301 """ Get id for this NSR"""
1302 return self
._nsr
_cfg
_msg
.id
1306 """ Name of this network service record """
1307 return self
._nsr
_cfg
_msg
.name
1310 def cloud_account_name(self
):
1311 return self
._nsr
_cfg
_msg
.cloud_account
1314 def om_datacenter_name(self
):
1315 if self
._nsr
_cfg
_msg
.has_field('om_datacenter'):
1316 return self
._nsr
_cfg
_msg
.om_datacenter
1321 """State of this NetworkServiceRecord"""
1322 return self
._op
_status
.state
1326 """ Is this NSR active ?"""
1327 return True if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
else False
1331 """ VLRs associated with this NSR"""
1336 """ VNFRs associated with this NSR"""
1341 """ VNFFGRs associated with this NSR"""
1342 return self
._vnffgrs
1345 def scaling_groups(self
):
1346 """ Scaling groups associated with this NSR """
1347 return self
._scaling
_groups
1350 def param_pools(self
):
1351 """ Parameter value pools associated with this NSR"""
1352 return self
._param
_pools
1355 def nsr_cfg_msg(self
):
1356 return self
._nsr
_cfg
_msg
1359 def nsr_cfg_msg(self
, msg
):
1360 self
._nsr
_cfg
_msg
= msg
1364 """ NSD Protobuf for this NSR """
1365 if self
._nsd
is not None:
1367 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
1372 """ NSD ID for this NSR """
1373 return self
.nsd_msg
.id
1377 ''' Get a new job id for config primitive'''
1382 def config_status(self
):
1383 """ Config status for NSR """
1384 return self
._config
_status
1386 def resolve_placement_group_cloud_construct(self
, input_group
):
1388 Returns the cloud specific construct for placement group
1390 copy_dict
= ['name', 'requirement', 'strategy']
1392 for group_info
in self
._nsr
_cfg
_msg
.nsd_placement_group_maps
:
1393 if group_info
.placement_group_ref
== input_group
.name
:
1394 group
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_PlacementGroupsInfo()
1395 group_dict
= {k
:v
for k
,v
in
1396 group_info
.as_dict().items() if k
!= 'placement_group_ref'}
1397 for param
in copy_dict
:
1398 group_dict
.update({param
: getattr(input_group
, param
)})
1399 group
.from_dict(group_dict
)
1405 return "NSR(name={}, nsd_id={}, cloud_account={})".format(
1406 self
.name
, self
.nsd_id
, self
.cloud_account_name
1409 def _get_vnfd(self
, vnfd_id
, config_xact
):
1410 """ Fetch vnfd msg for the passed vnfd id """
1411 return self
._nsm
.get_vnfd(vnfd_id
, config_xact
)
1413 def _get_vnfd_cloud_account(self
, vnfd_member_index
):
1414 """ Fetch Cloud Account for the passed vnfd id """
1415 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1416 vim_accounts
= [(vnf
.cloud_account
,vnf
.om_datacenter
) for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map \
1417 if vnfd_member_index
== vnf
.member_vnf_index_ref
]
1418 if vim_accounts
and vim_accounts
[0]:
1419 return vim_accounts
[0]
1420 return (self
.cloud_account_name
,self
.om_datacenter_name
)
1422 def _get_constituent_vnfd_msg(self
, vnf_index
):
1423 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1424 if const_vnfd
.member_vnf_index
== vnf_index
:
1427 raise ValueError("Constituent VNF index %s not found" % vnf_index
)
1429 def record_event(self
, evt
, evt_desc
, evt_details
=None, state
=None):
1430 """ Record an event """
1431 self
._op
_status
.record_event(evt
, evt_desc
, evt_details
)
1432 if state
is not None:
1433 self
.set_state(state
)
1435 def scaling_trigger_str(self
, trigger
):
1436 SCALING_TRIGGER_STRS
= {
1437 NsdYang
.ScalingTrigger
.PRE_SCALE_IN
: 'pre-scale-in',
1438 NsdYang
.ScalingTrigger
.POST_SCALE_IN
: 'post-scale-in',
1439 NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
: 'pre-scale-out',
1440 NsdYang
.ScalingTrigger
.POST_SCALE_OUT
: 'post-scale-out',
1443 return SCALING_TRIGGER_STRS
[trigger
]
1444 except Exception as e
:
1445 self
._log
.error("Scaling trigger mapping error for {} : {}".
1447 self
._log
.exception(e
)
1448 return "Unknown trigger"
1451 def instantiate_vls(self
):
1453 This function instantiates VLs for every VL in this Network Service
1455 self
._log
.debug("Instantiating %d VLs in NSD id %s", len(self
._vlrs
),
1457 for vlr
in self
._vlrs
:
1458 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1459 vlr
.state
= VlRecordState
.ACTIVE
1460 self
.vlr_uptime_tasks
[vlr
.id] = self
._loop
.create_task(self
.vlr_uptime_update(vlr
))
1463 def vlr_uptime_update(self
, vlr
):
1466 vlr_
= RwVlrYang
.YangData_Vlr_VlrCatalog_Vlr
.from_dict({'id': vlr
.id})
1468 vlr_
.uptime
= int(time
.time()) - vlr
._create
_time
1469 yield from self
._vlr
_handler
.update(None, VirtualLinkRecord
.vlr_xpath(vlr
), vlr_
)
1470 yield from asyncio
.sleep(2, loop
=self
._loop
)
1471 except asyncio
.CancelledError
:
1472 self
._log
.debug("Received cancellation request for vlr_uptime_update task")
1473 yield from self
._vlr
_handler
.delete(None, VirtualLinkRecord
.vlr_xpath(vlr
))
1477 def create(self
, config_xact
):
1478 """ Create this network service"""
1479 # Create virtual links for all the external vnf
1480 # connection points in this NS
1481 yield from self
.create_vls()
1483 # Create VNFs in this network service
1484 yield from self
.create_vnfs(config_xact
)
1486 # Create VNFFG for network service
1487 self
.create_vnffgs()
1489 # Create Scaling Groups for each scaling group in NSD
1490 self
.create_scaling_groups()
1492 # Create Parameter Pools
1493 self
.create_param_pools()
1496 def apply_scale_group_config_script(self
, script
, group
, scale_instance
, trigger
, vnfrs
=None):
1497 """ Apply config based on script for scale group """
1500 def add_vnfrs_data(vnfrs_list
):
1501 """ Add as a dict each of the VNFRs data """
1503 for vnfr
in vnfrs_list
:
1504 self
._log
.debug("Add VNFR {} data".format(vnfr
))
1506 vnfr_data
['name'] = vnfr
.name
1507 if trigger
in [NsdYang
.ScalingTrigger
.PRE_SCALE_IN
, NsdYang
.ScalingTrigger
.POST_SCALE_OUT
]:
1508 # Get VNF management and other IPs, etc
1509 opdata
= yield from self
.fetch_vnfr(vnfr
.xpath
)
1510 self
._log
.debug("VNFR {} op data: {}".format(vnfr
.name
, opdata
))
1512 vnfr_data
['rw_mgmt_ip'] = opdata
.mgmt_interface
.ip_address
1513 vnfr_data
['rw_mgmt_port'] = opdata
.mgmt_interface
.port
1514 except Exception as e
:
1515 self
._log
.error("Unable to get management IP for vnfr {}:{}".
1516 format(vnfr
.name
, e
))
1519 vnfr_data
['connection_points'] = []
1520 for cp
in opdata
.connection_point
:
1522 con_pt
['name'] = cp
.name
1523 con_pt
['ip_address'] = cp
.ip_address
1524 vnfr_data
['connection_points'].append(con_pt
)
1525 except Exception as e
:
1526 self
._log
.error("Exception getting connections points for VNFR {}: {}".
1527 format(vnfr
.name
, e
))
1529 vnfrs_data
.append(vnfr_data
)
1530 self
._log
.debug("VNFRs data: {}".format(vnfrs_data
))
1534 def add_nsr_data(nsr
):
1536 nsr_data
['name'] = nsr
.name
1539 if script
is None or len(script
) == 0:
1540 self
._log
.error("Script not provided for scale group config: {}".format(group
.name
))
1543 if script
[0] == '/':
1546 path
= os
.path
.join(os
.environ
['RIFT_INSTALL'], "usr/bin", script
)
1547 if not os
.path
.exists(path
):
1548 self
._log
.error("Config faled for scale group {}: Script does not exist at {}".
1549 format(group
.name
, path
))
1552 # Build a YAML file with all parameters for the script to execute
1553 # The data consists of 5 sections
1555 # 2. Scale group config
1556 # 3. VNFRs in the scale group
1557 # 4. VNFRs outside scale group
1560 data
['trigger'] = group
.trigger_map(trigger
)
1561 data
['config'] = group
.group_msg
.as_dict()
1564 data
["vnfrs_in_group"] = yield from add_vnfrs_data(vnfrs
)
1566 data
["vnfrs_in_group"] = yield from add_vnfrs_data(scale_instance
.vnfrs
)
1568 data
["vnfrs_others"] = yield from add_vnfrs_data(self
.vnfrs
.values())
1569 data
["nsr"] = add_nsr_data(self
)
1572 with tempfile
.NamedTemporaryFile(delete
=False) as tmp_file
:
1573 tmp_file
.write(yaml
.dump(data
, default_flow_style
=True)
1576 self
._log
.debug("Creating a temp file: {} with input data: {}".
1577 format(tmp_file
.name
, data
))
1579 cmd
= "{} {}".format(path
, tmp_file
.name
)
1580 self
._log
.debug("Running the CMD: {}".format(cmd
))
1581 proc
= yield from asyncio
.create_subprocess_shell(cmd
, loop
=self
._loop
)
1582 rc
= yield from proc
.wait()
1584 self
._log
.error("The script {} for scale group {} config returned: {}".
1585 format(script
, group
.name
, rc
))
1593 def apply_scaling_group_config(self
, trigger
, group
, scale_instance
, vnfrs
=None):
1594 """ Apply the config for the scaling group based on trigger """
1595 if group
is None or scale_instance
is None:
1599 def update_config_status(success
=True, err_msg
=None):
1600 self
._log
.debug("Update %s config status to %r : %s",
1601 scale_instance
, success
, err_msg
)
1602 if (scale_instance
.config_status
== "failed"):
1603 # Do not update the config status if it is already in failed state
1606 if scale_instance
.config_status
== "configured":
1607 # Update only to failed state an already configured scale instance
1609 scale_instance
.config_status
= "failed"
1610 scale_instance
.config_err_msg
= err_msg
1611 yield from self
.update_state()
1613 # We are in configuring state
1614 # Only after post scale out mark instance as configured
1615 if trigger
== NsdYang
.ScalingTrigger
.POST_SCALE_OUT
:
1617 scale_instance
.config_status
= "configured"
1619 scale_instance
.config_status
= "failed"
1620 scale_instance
.config_err_msg
= err_msg
1621 yield from self
.update_state()
1623 config
= group
.trigger_config(trigger
)
1627 self
._log
.debug("Scaling group {} config: {}".format(group
.name
, config
))
1628 if config
.has_field("ns_config_primitive_name_ref"):
1629 config_name
= config
.ns_config_primitive_name_ref
1630 nsd_msg
= self
.nsd_msg
1631 config_primitive
= None
1632 for ns_cfg_prim
in nsd_msg
.service_primitive
:
1633 if ns_cfg_prim
.name
== config_name
:
1634 config_primitive
= ns_cfg_prim
1637 if config_primitive
is None:
1638 raise ValueError("Could not find ns_cfg_prim %s in nsr %s" % (config_name
, self
.name
))
1640 self
._log
.debug("Scaling group {} config primitive: {}".format(group
.name
, config_primitive
))
1641 if config_primitive
.has_field("user_defined_script"):
1642 rc
= yield from self
.apply_scale_group_config_script(config_primitive
.user_defined_script
,
1643 group
, scale_instance
, trigger
, vnfrs
)
1646 err_msg
= "Failed config for trigger {} using config script '{}'". \
1647 format(self
.scaling_trigger_str(trigger
),
1648 config_primitive
.user_defined_script
)
1649 yield from update_config_status(success
=rc
, err_msg
=err_msg
)
1652 err_msg
= "Failed config for trigger {} as config script is not specified". \
1653 format(self
.scaling_trigger_str(trigger
))
1654 yield from update_config_status(success
=False, err_msg
=err_msg
)
1655 raise NotImplementedError("Only script based config support for scale group for now: {}".
1658 err_msg
= "Failed config for trigger {} as config primitive is not specified".\
1659 format(self
.scaling_trigger_str(trigger
))
1660 yield from update_config_status(success
=False, err_msg
=err_msg
)
1661 self
._log
.error("Config primitive not specified for config action in scale group %s" %
1665 def create_scaling_groups(self
):
1666 """ This function creates a NSScalingGroup for every scaling
1667 group defined in he NSD"""
1669 for scaling_group_msg
in self
.nsd_msg
.scaling_group_descriptor
:
1670 self
._log
.debug("Found scaling_group %s in nsr id %s",
1671 scaling_group_msg
.name
, self
.id)
1673 group_record
= scale_group
.ScalingGroup(
1678 self
._scaling
_groups
[group_record
.name
] = group_record
1681 def create_scale_group_instance(self
, group_name
, index
, config_xact
, is_default
=False):
1682 group
= self
._scaling
_groups
[group_name
]
1683 scale_instance
= group
.create_instance(index
, is_default
)
1687 self
._log
.debug("Creating %u VNFs associated with NS id %s scaling group %s",
1688 len(self
.nsd_msg
.constituent_vnfd
), self
.id, self
)
1691 for vnf_index
, count
in group
.vnf_index_count_map
.items():
1692 const_vnfd_msg
= self
._get
_constituent
_vnfd
_msg
(vnf_index
)
1693 vnfd_msg
= self
._get
_vnfd
(const_vnfd_msg
.vnfd_id_ref
, config_xact
)
1695 cloud_account_name
, om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd_msg
.member_vnf_index
)
1696 if cloud_account_name
is None:
1697 cloud_account_name
= self
.cloud_account_name
1698 for _
in range(count
):
1699 vnfr
= yield from self
.create_vnf_record(vnfd_msg
, const_vnfd_msg
, cloud_account_name
, om_datacenter_name
, group_name
, index
)
1700 scale_instance
.add_vnfr(vnfr
)
1706 def instantiate_instance():
1707 self
._log
.debug("Creating %s VNFRS", scale_instance
)
1708 vnfrs
= yield from create_vnfs()
1709 yield from self
.publish()
1711 self
._log
.debug("Instantiating %s VNFRS for %s", len(vnfrs
), scale_instance
)
1712 scale_instance
.operational_status
= "vnf_init_phase"
1713 yield from self
.update_state()
1716 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
,
1717 group
, scale_instance
, vnfrs
)
1719 self
._log
.error("Pre scale out config for scale group {} ({}) failed".
1720 format(group
.name
, index
))
1721 scale_instance
.operational_status
= "failed"
1723 yield from self
.instantiate_vnfs(vnfrs
)
1725 except Exception as e
:
1726 self
._log
.exception("Failed to begin instantiatiation of vnfs for scale group {}: {}".
1727 format(group
.name
, e
))
1728 self
._log
.exception(e
)
1729 scale_instance
.operational_status
= "failed"
1731 yield from self
.update_state()
1733 yield from instantiate_instance()
1736 def delete_scale_group_instance(self
, group_name
, index
):
1737 group
= self
._scaling
_groups
[group_name
]
1738 scale_instance
= group
.get_instance(index
)
1739 if scale_instance
.is_default
:
1740 raise ScalingOperationError("Cannot terminate a default scaling group instance")
1742 scale_instance
.operational_status
= "terminate"
1743 yield from self
.update_state()
1746 def terminate_instance():
1747 self
._log
.debug("Terminating %s VNFRS" % scale_instance
)
1748 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_IN
,
1749 group
, scale_instance
)
1751 self
._log
.error("Pre scale in config for scale group {} ({}) failed".
1752 format(group
.name
, index
))
1754 # Going ahead with terminate, even if there is an error in pre-scale-in config
1755 # as this could be result of scale out failure and we need to cleanup this group
1756 yield from self
.terminate_vnfrs(scale_instance
.vnfrs
)
1757 group
.delete_instance(index
)
1759 scale_instance
.operational_status
= "vnf_terminate_phase"
1760 yield from self
.update_state()
1762 yield from terminate_instance()
1765 def _update_scale_group_instances_status(self
):
1767 def post_scale_out_task(group
, instance
):
1768 # Apply post scale out config once all VNFRs are active
1769 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_OUT
,
1771 instance
.operational_status
= "running"
1773 self
._log
.debug("Scale out for group {} and instance {} succeeded".
1774 format(group
.name
, instance
.instance_id
))
1776 self
._log
.error("Post scale out config for scale group {} ({}) failed".
1777 format(group
.name
, instance
.instance_id
))
1779 yield from self
.update_state()
1781 group_instances
= {group
: group
.instances
for group
in self
._scaling
_groups
.values()}
1782 for group
, instances
in group_instances
.items():
1783 self
._log
.debug("Updating %s instance status", group
)
1784 for instance
in instances
:
1785 instance_vnf_state_list
= [vnfr
.state
for vnfr
in instance
.vnfrs
]
1786 self
._log
.debug("Got vnfr instance states: %s", instance_vnf_state_list
)
1787 if instance
.operational_status
== "vnf_init_phase":
1788 if all([state
== VnfRecordState
.ACTIVE
for state
in instance_vnf_state_list
]):
1789 instance
.operational_status
= "running"
1791 # Create a task for post scale out to allow us to sleep before attempting
1792 # to configure newly created VM's
1793 self
._loop
.create_task(post_scale_out_task(group
, instance
))
1795 elif any([state
== VnfRecordState
.FAILED
for state
in instance_vnf_state_list
]):
1796 self
._log
.debug("Scale out for group {} and instance {} failed".
1797 format(group
.name
, instance
.instance_id
))
1798 instance
.operational_status
= "failed"
1800 elif instance
.operational_status
== "vnf_terminate_phase":
1801 if all([state
== VnfRecordState
.TERMINATED
for state
in instance_vnf_state_list
]):
1802 instance
.operational_status
= "terminated"
1803 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_IN
,
1806 self
._log
.debug("Scale in for group {} and instance {} succeeded".
1807 format(group
.name
, instance
.instance_id
))
1809 self
._log
.error("Post scale in config for scale group {} ({}) failed".
1810 format(group
.name
, instance
.instance_id
))
1812 def create_vnffgs(self
):
1813 """ This function creates VNFFGs for every VNFFG in the NSD
1814 associated with this NSR"""
1816 for vnffgd
in self
.nsd_msg
.vnffgd
:
1817 self
._log
.debug("Found vnffgd %s in nsr id %s", vnffgd
, self
.id)
1818 vnffgr
= VnffgRecord(self
._dts
,
1821 self
._nsm
._vnffgmgr
,
1825 self
._sdn
_account
_name
1827 self
._vnffgrs
[vnffgr
.id] = vnffgr
1829 def resolve_vld_ip_profile(self
, nsd_msg
, vld
):
1830 self
._log
.debug("Receieved ip profile ref is %s",vld
.ip_profile_ref
)
1831 if not vld
.has_field('ip_profile_ref'):
1833 profile
= [profile
for profile
in nsd_msg
.ip_profiles
if profile
.name
== vld
.ip_profile_ref
]
1834 return profile
[0] if profile
else None
1837 def _create_vls(self
, vld
, cloud_account
,om_datacenter
):
1838 """Create a VLR in the cloud account specified using the given VLD
1842 cloud_account : Cloud account name
1847 vlr
= yield from VirtualLinkRecord
.create_record(
1855 self
.resolve_vld_ip_profile(self
.nsd_msg
, vld
),
1857 restart_mode
=self
.restart_mode
)
1861 def _extract_cloud_accounts_for_vl(self
, vld
):
1863 Extracts the list of cloud accounts from the NS Config obj
1866 1. Cloud accounts based connection point (vnf_cloud_account_map)
1868 vld : VLD yang object
1873 cloud_account_list
= []
1875 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1876 # Handle case where cloud_account is None
1878 for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1879 if vnf
.cloud_account
is not None or vnf
.om_datacenter
is not None:
1880 vnf_cloud_map
[vnf
.member_vnf_index_ref
] = (vnf
.cloud_account
,vnf
.om_datacenter
)
1882 for vnfc
in vld
.vnfd_connection_point_ref
:
1883 cloud_account
= vnf_cloud_map
.get(
1884 vnfc
.member_vnf_index_ref
,
1885 (self
.cloud_account_name
,self
.om_datacenter_name
))
1887 cloud_account_list
.append(cloud_account
)
1889 if self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1890 for vld_map
in self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1891 if vld_map
.vld_id_ref
== vld
.id:
1892 for cloud_account
in vld_map
.cloud_accounts
:
1893 cloud_account_list
.extend((cloud_account
,None))
1894 for om_datacenter
in vld_map
.om_datacenters
:
1895 cloud_account_list
.extend((None,om_datacenter
))
1897 # If no config has been provided then fall-back to the default
1899 if not cloud_account_list
:
1900 cloud_account_list
= [(self
.cloud_account_name
,self
.om_datacenter_name
)]
1902 self
._log
.debug("VL {} cloud accounts: {}".
1903 format(vld
.name
, cloud_account_list
))
1904 return set(cloud_account_list
)
1907 def create_vls(self
):
1908 """ This function creates VLs for every VLD in the NSD
1909 associated with this NSR"""
1910 for vld
in self
.nsd_msg
.vld
:
1912 self
._log
.debug("Found vld %s in nsr id %s", vld
, self
.id)
1913 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1914 for cloud_account
,om_datacenter
in cloud_account_list
:
1915 vlr
= yield from self
._create
_vls
(vld
, cloud_account
,om_datacenter
)
1916 self
._vlrs
.append(vlr
)
1920 def create_vl_instance(self
, vld
):
1921 self
._log
.debug("Create VL for {}: {}".format(self
.id, vld
.as_dict()))
1922 # Check if the VL is already present
1924 for vl
in self
._vlrs
:
1925 if vl
.vld_msg
.id == vld
.id:
1926 self
._log
.debug("The VLD %s already in NSR %s as VLR %s with status %s",
1927 vld
.id, self
.id, vl
.id, vl
.state
)
1929 if vlr
.state
!= VlRecordState
.TERMINATED
:
1930 err_msg
= "VLR for VL %s in NSR %s already instantiated", \
1932 self
._log
.error(err_msg
)
1933 raise NsrVlUpdateError(err_msg
)
1937 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1938 for account
,om_datacenter
in cloud_account_list
:
1939 vlr
= yield from self
._create
_vls
(vld
, account
,om_datacenter
)
1940 self
._vlrs
.append(vlr
)
1942 vlr
.state
= VlRecordState
.INSTANTIATION_PENDING
1943 yield from self
.update_state()
1946 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1947 vlr
.state
= VlRecordState
.ACTIVE
1949 except Exception as e
:
1950 err_msg
= "Error instantiating VL for NSR {} and VLD {}: {}". \
1951 format(self
.id, vld
.id, e
)
1952 self
._log
.error(err_msg
)
1953 self
._log
.exception(e
)
1954 vlr
.state
= VlRecordState
.FAILED
1956 yield from self
.update_state()
1959 def delete_vl_instance(self
, vld
):
1960 for vlr
in self
._vlrs
:
1961 if vlr
.vld_msg
.id == vld
.id:
1962 self
._log
.debug("Found VLR %s for VLD %s in NSR %s",
1963 vlr
.id, vld
.id, self
.id)
1964 vlr
.state
= VlRecordState
.TERMINATE_PENDING
1965 yield from self
.update_state()
1968 yield from self
.nsm_plugin
.terminate_vl(vlr
)
1969 vlr
.state
= VlRecordState
.TERMINATED
1970 self
._vlrs
.remove(vlr
)
1972 except Exception as e
:
1973 err_msg
= "Error terminating VL for NSR {} and VLD {}: {}". \
1974 format(self
.id, vld
.id, e
)
1975 self
._log
.error(err_msg
)
1976 self
._log
.exception(e
)
1977 vlr
.state
= VlRecordState
.FAILED
1979 yield from self
.update_state()
1983 def create_vnfs(self
, config_xact
):
1985 This function creates VNFs for every VNF in the NSD
1986 associated with this NSR
1988 self
._log
.debug("Creating %u VNFs associated with this NS id %s",
1989 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
1991 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1992 if not const_vnfd
.start_by_default
:
1993 self
._log
.debug("start_by_default set to False in constituent VNF (%s). Skipping start.",
1994 const_vnfd
.member_vnf_index
)
1997 vnfd_msg
= self
._get
_vnfd
(const_vnfd
.vnfd_id_ref
, config_xact
)
1998 cloud_account_name
,om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd
.member_vnf_index
)
1999 if cloud_account_name
is None:
2000 cloud_account_name
= self
.cloud_account_name
2001 yield from self
.create_vnf_record(vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
)
2004 def get_placement_groups(self
, vnfd_msg
, const_vnfd
):
2005 placement_groups
= []
2006 for group
in self
.nsd_msg
.placement_groups
:
2007 for member_vnfd
in group
.member_vnfd
:
2008 if (member_vnfd
.vnfd_id_ref
== vnfd_msg
.id) and \
2009 (member_vnfd
.member_vnf_index_ref
== const_vnfd
.member_vnf_index
):
2010 group_info
= self
.resolve_placement_group_cloud_construct(group
)
2011 if group_info
is None:
2012 self
._log
.error("Could not resolve cloud-construct for placement group: %s", group
.name
)
2013 ### raise PlacementGroupError("Could not resolve cloud-construct for placement group: {}".format(group.name))
2015 self
._log
.info("Successfully resolved cloud construct for placement group: %s for VNF: %s (Member Index: %s)",
2018 const_vnfd
.member_vnf_index
)
2019 placement_groups
.append(group_info
)
2020 return placement_groups
2023 def create_vnf_record(self
, vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
, group_name
=None, group_instance_id
=None):
2024 # Fetch the VNFD associated with this VNF
2025 placement_groups
= self
.get_placement_groups(vnfd_msg
, const_vnfd
)
2026 self
._log
.info("Cloud Account for VNF %d is %s",const_vnfd
.member_vnf_index
,cloud_account_name
)
2027 self
._log
.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s",
2029 const_vnfd
.member_vnf_index
,
2030 [ group
.name
for group
in placement_groups
])
2031 vnfr
= yield from VirtualNetworkFunctionRecord
.create_record(self
._dts
,
2044 restart_mode
=self
.restart_mode
,
2046 if vnfr
.id in self
._vnfrs
:
2047 err
= "VNF with VNFR id %s already in vnf list" % (vnfr
.id,)
2048 raise NetworkServiceRecordError(err
)
2050 self
._vnfrs
[vnfr
.id] = vnfr
2051 self
._nsm
.vnfrs
[vnfr
.id] = vnfr
2053 yield from vnfr
.set_config_status(NsrYang
.ConfigStates
.INIT
)
2055 self
._log
.debug("Added VNFR %s to NSM VNFR list with id %s",
2061 def create_param_pools(self
):
2062 for param_pool
in self
.nsd_msg
.parameter_pool
:
2063 self
._log
.debug("Found parameter pool %s in nsr id %s", param_pool
, self
.id)
2065 start_value
= param_pool
.range.start_value
2066 end_value
= param_pool
.range.end_value
2067 if end_value
< start_value
:
2068 raise NetworkServiceRecordError(
2069 "Parameter pool %s has invalid range (start: {}, end: {})".format(
2070 start_value
, end_value
2074 self
._param
_pools
[param_pool
.name
] = config_value_pool
.ParameterValuePool(
2077 range(start_value
, end_value
)
2081 def fetch_vnfr(self
, vnfr_path
):
2082 """ Fetch VNFR record """
2084 self
._log
.debug("Fetching VNFR with key %s while instantiating %s",
2086 res_iter
= yield from self
._dts
.query_read(vnfr_path
, rwdts
.XactFlag
.MERGE
)
2088 for ent
in res_iter
:
2089 res
= yield from ent
2095 def instantiate_vnfs(self
, vnfrs
):
2097 This function instantiates VNFs for every VNF in this Network Service
2099 self
._log
.debug("Instantiating %u VNFs in NS %s", len(vnfrs
), self
.id)
2101 self
._log
.debug("Instantiating VNF: %s in NS %s", vnf
, self
.id)
2102 yield from self
.nsm_plugin
.instantiate_vnf(self
, vnf
)
2105 def instantiate_vnffgs(self
):
2107 This function instantiates VNFFGs for every VNFFG in this Network Service
2109 self
._log
.debug("Instantiating %u VNFFGs in NS %s",
2110 len(self
.nsd_msg
.vnffgd
), self
.id)
2111 for _
, vnfr
in self
.vnfrs
.items():
2112 while vnfr
.state
in [VnfRecordState
.INSTANTIATION_PENDING
, VnfRecordState
.INIT
]:
2113 self
._log
.debug("Received vnfr state for vnfr %s is %s; retrying",vnfr
.name
,vnfr
.state
)
2114 yield from asyncio
.sleep(2, loop
=self
._loop
)
2115 if vnfr
.state
== VnfRecordState
.ACTIVE
:
2116 self
._log
.debug("Received vnfr state for vnfr %s is %s ",vnfr
.name
,vnfr
.state
)
2119 self
._log
.debug("Received vnfr state for vnfr %s is %s; failing vnffg creation",vnfr
.name
,vnfr
.state
)
2120 self
._vnffgr
_state
= VnffgRecordState
.FAILED
2123 self
._log
.info("Waiting for 90 seconds for VMs to come up")
2124 yield from asyncio
.sleep(90, loop
=self
._loop
)
2125 self
._log
.info("Starting VNFFG orchestration")
2126 for vnffg
in self
._vnffgrs
.values():
2127 self
._log
.debug("Instantiating VNFFG: %s in NS %s", vnffg
, self
.id)
2128 yield from vnffg
.instantiate()
2131 def instantiate_scaling_instances(self
, config_xact
):
2132 """ Instantiate any default scaling instances in this Network Service """
2133 for group
in self
._scaling
_groups
.values():
2134 for i
in range(group
.min_instance_count
):
2135 self
._log
.debug("Instantiating %s default scaling instance %s", group
, i
)
2136 yield from self
.create_scale_group_instance(
2137 group
.name
, i
, config_xact
, is_default
=True
2140 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2141 if group_msg
.scaling_group_name_ref
!= group
.name
:
2144 for instance
in group_msg
.instance
:
2145 self
._log
.debug("Reloading %s scaling instance %s", group_msg
, instance
.id)
2146 yield from self
.create_scale_group_instance(
2147 group
.name
, instance
.id, config_xact
, is_default
=False
2150 def has_scaling_instances(self
):
2151 """ Return boolean indicating if the network service has default scaling groups """
2152 for group
in self
._scaling
_groups
.values():
2153 if group
.min_instance_count
> 0:
2156 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2157 if len(group_msg
.instance
) > 0:
2164 """ This function publishes this NSR """
2165 self
._nsr
_msg
= self
.create_msg()
2167 self
._log
.debug("Publishing the NSR with xpath %s and nsr %s",
2171 if self
._debug
_running
:
2172 self
._log
.debug("Publishing NSR in RUNNING state!")
2175 with self
._dts
.transaction() as xact
:
2176 yield from self
._nsm
.nsr_handler
.update(xact
, self
.nsr_xpath
, self
._nsr
_msg
)
2177 if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
:
2178 self
._debug
_running
= True
2181 def unpublish(self
, xact
):
2182 """ Unpublish this NSR object """
2183 self
._log
.debug("Unpublishing Network service id %s", self
.id)
2184 yield from self
._nsm
.nsr_handler
.delete(xact
, self
.nsr_xpath
)
2187 def nsr_xpath(self
):
2188 """ Returns the xpath associated with this NSR """
2190 "D,/nsr:ns-instance-opdata" +
2191 "/nsr:nsr[nsr:ns-instance-config-ref = '{}']"
2195 def xpath_from_nsr(nsr
):
2196 """ Returns the xpath associated with this NSR op data"""
2197 return (NetworkServiceRecord
.XPATH
+
2198 "[nsr:ns-instance-config-ref = '{}']").format(nsr
.id)
2201 def nsd_xpath(self
):
2202 """ Return NSD config xpath."""
2204 "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}']"
2205 ).format(self
.nsd_id
)
2208 def instantiate(self
, config_xact
):
2209 """"Instantiates a NetworkServiceRecord.
2211 This function instantiates a Network service
2212 which involves the following steps,
2214 * Instantiate every VL in NSD by sending create VLR request to DTS.
2215 * Instantiate every VNF in NSD by sending create VNF reuqest to DTS.
2216 * Publish the NSR details to DTS
2219 nsr: The NSR configuration request containing nsr-id and nsd
2220 config_xact: The configuration transaction which initiated the instatiation
2223 NetworkServiceRecordError if the NSR creation fails
2229 self
._log
.debug("Instantiating NS - %s xact - %s", self
, config_xact
)
2231 # Move the state to INIITALIZING
2232 self
.set_state(NetworkServiceRecordState
.INIT
)
2234 event_descr
= "Instantiation Request Received NSR Id:%s" % self
.id
2235 self
.record_event("instantiating", event_descr
)
2238 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
2241 # Update ref count if nsd present in catalog
2242 self
._nsm
.get_nsd_ref(self
.nsd_id
)
2244 except NetworkServiceDescriptorError
:
2245 # This could be an NSD not in the nsd-catalog
2248 # Merge any config and initial config primitive values
2249 self
.config_store
.merge_nsd_config(self
.nsd_msg
)
2250 self
._log
.debug("Merged NSD: {}".format(self
.nsd_msg
.as_dict()))
2252 event_descr
= "Fetched NSD with descriptor id %s" % self
.nsd_id
2253 self
.record_event("nsd-fetched", event_descr
)
2255 if self
._nsd
is None:
2256 msg
= "Failed to fetch NSD with nsd-id [%s] for nsr-id %s"
2257 self
._log
.debug(msg
, self
.nsd_id
, self
.id)
2258 raise NetworkServiceRecordError(self
)
2260 self
._log
.debug("Got nsd result %s", self
._nsd
)
2262 # Substitute any input parameters
2263 self
.substitute_input_parameters(self
._nsd
, self
._nsr
_cfg
_msg
)
2266 yield from self
.create(config_xact
)
2268 # Publish the NSR to DTS
2269 yield from self
.publish()
2272 def do_instantiate():
2274 Instantiate network service
2276 self
._log
.debug("Instantiating VLs nsr id [%s] nsd id [%s]",
2277 self
.id, self
.nsd_id
)
2279 # instantiate the VLs
2280 event_descr
= ("Instantiating %s external VLs for NSR id %s" %
2281 (len(self
.nsd_msg
.vld
), self
.id))
2282 self
.record_event("begin-external-vls-instantiation", event_descr
)
2284 self
.set_state(NetworkServiceRecordState
.VL_INIT_PHASE
)
2286 yield from self
.instantiate_vls()
2288 # Publish the NSR to DTS
2289 yield from self
.publish()
2291 event_descr
= ("Finished instantiating %s external VLs for NSR id %s" %
2292 (len(self
.nsd_msg
.vld
), self
.id))
2293 self
.record_event("end-external-vls-instantiation", event_descr
)
2295 self
.set_state(NetworkServiceRecordState
.VNF_INIT_PHASE
)
2297 self
._log
.debug("Instantiating VNFs ...... nsr[%s], nsd[%s]",
2298 self
.id, self
.nsd_id
)
2300 # instantiate the VNFs
2301 event_descr
= ("Instantiating %s VNFS for NSR id %s" %
2302 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2304 self
.record_event("begin-vnf-instantiation", event_descr
)
2306 yield from self
.instantiate_vnfs(self
._vnfrs
.values())
2308 self
._log
.debug(" Finished instantiating %d VNFs for NSR id %s",
2309 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
2311 event_descr
= ("Finished instantiating %s VNFs for NSR id %s" %
2312 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2313 self
.record_event("end-vnf-instantiation", event_descr
)
2315 if len(self
.vnffgrs
) > 0:
2316 #self.set_state(NetworkServiceRecordState.VNFFG_INIT_PHASE)
2317 event_descr
= ("Instantiating %s VNFFGS for NSR id %s" %
2318 (len(self
.nsd_msg
.vnffgd
), self
.id))
2320 self
.record_event("begin-vnffg-instantiation", event_descr
)
2322 yield from self
.instantiate_vnffgs()
2324 event_descr
= ("Finished instantiating %s VNFFGDs for NSR id %s" %
2325 (len(self
.nsd_msg
.vnffgd
), self
.id))
2326 self
.record_event("end-vnffg-instantiation", event_descr
)
2328 if self
.has_scaling_instances():
2329 event_descr
= ("Instantiating %s Scaling Groups for NSR id %s" %
2330 (len(self
._scaling
_groups
), self
.id))
2332 self
.record_event("begin-scaling-group-instantiation", event_descr
)
2333 yield from self
.instantiate_scaling_instances(config_xact
)
2334 self
.record_event("end-scaling-group-instantiation", event_descr
)
2336 # Give the plugin a chance to deploy the network service now that all
2337 # virtual links and vnfs are instantiated
2338 yield from self
.nsm_plugin
.deploy(self
._nsr
_msg
)
2340 self
._log
.debug("Publishing NSR...... nsr[%s], nsd[%s]",
2341 self
.id, self
.nsd_id
)
2343 # Publish the NSR to DTS
2344 yield from self
.publish()
2346 self
._log
.debug("Published NSR...... nsr[%s], nsd[%s]",
2347 self
.id, self
.nsd_id
)
2349 def on_instantiate_done(fut
):
2350 # If the do_instantiate fails, then publish NSR with failed result
2351 if fut
.exception() is not None:
2352 self
._log
.error("NSR instantiation failed for NSR id %s: %s", self
.id, str(fut
.exception()))
2353 self
._loop
.create_task(self
.instantiation_failed(failed_reason
=str(fut
.exception())))
2355 instantiate_task
= self
._loop
.create_task(do_instantiate())
2356 instantiate_task
.add_done_callback(on_instantiate_done
)
2359 def set_config_status(self
, status
, status_details
=None):
2360 if self
.config_status
!= status
:
2361 self
._log
.debug("Updating NSR {} status for {} to {}".
2362 format(self
.name
, self
.config_status
, status
))
2363 self
._config
_status
= status
2364 self
._config
_status
_details
= status_details
2366 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2367 self
.record_event("config-failed", "NS configuration failed",
2368 evt_details
=self
._config
_status
_details
)
2370 yield from self
.publish()
2373 def is_active(self
):
2374 """ This NS is active """
2375 self
.set_state(NetworkServiceRecordState
.RUNNING
)
2379 # Publish the NSR to DTS
2380 self
._log
.debug("Network service %s is active ", self
.id)
2381 self
._is
_active
= True
2383 event_descr
= "NSR in running state for NSR id %s" % self
.id
2384 self
.record_event("ns-running", event_descr
)
2386 yield from self
.publish()
2389 def instantiation_failed(self
, failed_reason
=None):
2390 """ The NS instantiation failed"""
2391 self
._log
.error("Network service id:%s, name:%s instantiation failed",
2393 self
.set_state(NetworkServiceRecordState
.FAILED
)
2395 event_descr
= "Instantiation of NS %s failed" % self
.id
2396 self
.record_event("ns-failed", event_descr
, evt_details
=failed_reason
)
2398 # Publish the NSR to DTS
2399 yield from self
.publish()
2402 def terminate_vnfrs(self
, vnfrs
):
2403 """ Terminate VNFRS in this network service """
2404 self
._log
.debug("Terminating VNFs in network service %s", self
.id)
2406 yield from self
.nsm_plugin
.terminate_vnf(vnfr
)
2409 def terminate(self
):
2410 """ Terminate a NetworkServiceRecord."""
2411 def terminate_vnffgrs():
2412 """ Terminate VNFFGRS in this network service """
2413 self
._log
.debug("Terminating VNFFGRs in network service %s", self
.id)
2414 for vnffgr
in self
.vnffgrs
.values():
2415 yield from vnffgr
.terminate()
2417 def terminate_vlrs():
2418 """ Terminate VLRs in this netork service """
2419 self
._log
.debug("Terminating VLs in network service %s", self
.id)
2420 for vlr
in self
.vlrs
:
2421 yield from self
.nsm_plugin
.terminate_vl(vlr
)
2422 vlr
.state
= VlRecordState
.TERMINATED
2423 if vlr
.id in self
.vlr_uptime_tasks
:
2424 self
.vlr_uptime_tasks
[vlr
.id].cancel()
2426 self
._log
.debug("Terminating network service id %s", self
.id)
2428 # Move the state to TERMINATE
2429 self
.set_state(NetworkServiceRecordState
.TERMINATE
)
2430 event_descr
= "Terminate being processed for NS Id:%s" % self
.id
2431 self
.record_event("terminate", event_descr
)
2433 # Move the state to VNF_TERMINATE_PHASE
2434 self
._log
.debug("Terminating VNFFGs in NS ID: %s", self
.id)
2435 self
.set_state(NetworkServiceRecordState
.VNFFG_TERMINATE_PHASE
)
2436 event_descr
= "Terminating VNFFGS in NS Id:%s" % self
.id
2437 self
.record_event("terminating-vnffgss", event_descr
)
2438 yield from terminate_vnffgrs()
2440 # Move the state to VNF_TERMINATE_PHASE
2441 self
.set_state(NetworkServiceRecordState
.VNF_TERMINATE_PHASE
)
2442 event_descr
= "Terminating VNFS in NS Id:%s" % self
.id
2443 self
.record_event("terminating-vnfs", event_descr
)
2444 yield from self
.terminate_vnfrs(self
.vnfrs
.values())
2446 # Move the state to VL_TERMINATE_PHASE
2447 self
.set_state(NetworkServiceRecordState
.VL_TERMINATE_PHASE
)
2448 event_descr
= "Terminating VLs in NS Id:%s" % self
.id
2449 self
.record_event("terminating-vls", event_descr
)
2450 yield from terminate_vlrs()
2452 yield from self
.nsm_plugin
.terminate_ns(self
)
2454 # Move the state to TERMINATED
2455 self
.set_state(NetworkServiceRecordState
.TERMINATED
)
2456 event_descr
= "Terminated NS Id:%s" % self
.id
2457 self
.record_event("terminated", event_descr
)
2460 """"Enable a NetworkServiceRecord."""
2464 """"Disable a NetworkServiceRecord."""
2467 def map_config_status(self
):
2468 self
._log
.debug("Config status for ns {} is {}".
2469 format(self
.name
, self
._config
_status
))
2470 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURING
:
2471 return 'configuring'
2472 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2476 def vl_phase_completed(self
):
2477 """ Are VLs created in this NS?"""
2478 return self
._vl
_phase
_completed
2480 def vnf_phase_completed(self
):
2481 """ Are VLs created in this NS?"""
2482 return self
._vnf
_phase
_completed
2484 def create_msg(self
):
2485 """ The network serice record as a message """
2486 nsr_dict
= {"ns_instance_config_ref": self
.id}
2487 nsr
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr
.from_dict(nsr_dict
)
2488 #nsr.cloud_account = self.cloud_account_name
2489 nsr
.sdn_account
= self
._sdn
_account
_name
2490 nsr
.name_ref
= self
.name
2491 nsr
.nsd_ref
= self
.nsd_id
2492 nsr
.nsd_name_ref
= self
.nsd_msg
.name
2493 nsr
.operational_events
= self
._op
_status
.msg
2494 nsr
.operational_status
= self
._op
_status
.yang_str()
2495 nsr
.config_status
= self
.map_config_status()
2496 nsr
.config_status_details
= self
._config
_status
_details
2497 nsr
.create_time
= self
._create
_time
2498 nsr
.uptime
= int(time
.time()) - self
._create
_time
2500 for cfg_prim
in self
.nsd_msg
.service_primitive
:
2501 cfg_prim
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_ServicePrimitive
.from_dict(
2503 nsr
.service_primitive
.append(cfg_prim
)
2505 for init_cfg
in self
.nsd_msg
.initial_config_primitive
:
2506 prim
= NsrYang
.NsrInitialConfigPrimitive
.from_dict(
2508 nsr
.initial_config_primitive
.append(prim
)
2510 if self
.vl_phase_completed():
2511 for vlr
in self
.vlrs
:
2512 nsr
.vlr
.append(vlr
.create_nsr_vlr_msg(self
.vnfrs
.values()))
2514 if self
.vnf_phase_completed():
2515 for vnfr_id
in self
.vnfrs
:
2516 nsr
.constituent_vnfr_ref
.append(self
.vnfrs
[vnfr_id
].const_vnfr_msg
)
2517 for vnffgr
in self
.vnffgrs
.values():
2518 nsr
.vnffgr
.append(vnffgr
.fetch_vnffgr())
2519 for scaling_group
in self
._scaling
_groups
.values():
2520 nsr
.scaling_group_record
.append(scaling_group
.create_record_msg())
2524 def all_vnfs_active(self
):
2525 """ Are all VNFS in this NS active? """
2526 for _
, vnfr
in self
.vnfrs
.items():
2527 if vnfr
.active
is not True:
2532 def update_state(self
):
2533 """ Re-evaluate this NS's state """
2534 curr_state
= self
._op
_status
.state
2536 if curr_state
== NetworkServiceRecordState
.TERMINATED
:
2537 self
._log
.debug("NS (%s) in terminated state, not updating state", self
.id)
2540 new_state
= NetworkServiceRecordState
.RUNNING
2541 self
._log
.info("Received update_state for nsr: %s, curr-state: %s",
2542 self
.id, curr_state
)
2544 # Check all the VNFRs are present
2545 for _
, vnfr
in self
.vnfrs
.items():
2546 if vnfr
.state
in [VnfRecordState
.ACTIVE
, VnfRecordState
.TERMINATED
]:
2548 elif vnfr
.state
== VnfRecordState
.FAILED
:
2549 if vnfr
._prev
_state
!= vnfr
.state
:
2550 event_descr
= "Instantiation of VNF %s failed" % vnfr
.id
2551 event_error_details
= vnfr
.state_failed_reason
2552 self
.record_event("vnf-failed", event_descr
, evt_details
=event_error_details
)
2553 vnfr
.set_state(VnfRecordState
.FAILED
)
2555 self
._log
.info("VNF state did not change, curr=%s, prev=%s",
2556 vnfr
.state
, vnfr
._prev
_state
)
2557 new_state
= NetworkServiceRecordState
.FAILED
2560 self
._log
.info("VNF %s in NSR %s is still not active; current state is: %s",
2561 vnfr
.id, self
.id, vnfr
.state
)
2562 new_state
= curr_state
2564 # If new state is RUNNING; check all VLs
2565 if new_state
== NetworkServiceRecordState
.RUNNING
:
2566 for vl
in self
.vlrs
:
2568 if vl
.state
in [VlRecordState
.ACTIVE
, VlRecordState
.TERMINATED
]:
2570 elif vl
.state
== VlRecordState
.FAILED
:
2571 if vl
.prev_state
!= vl
.state
:
2572 event_descr
= "Instantiation of VL %s failed" % vl
.id
2573 event_error_details
= vl
.state_failed_reason
2574 self
.record_event("vl-failed", event_descr
, evt_details
=event_error_details
)
2575 vl
.prev_state
= vl
.state
2577 self
._log
.debug("VL %s already in failed state")
2579 if vl
.state
in [VlRecordState
.INSTANTIATION_PENDING
, VlRecordState
.INIT
]:
2580 new_state
= NetworkServiceRecordState
.VL_INSTANTIATE
2583 if vl
.state
in [VlRecordState
.TERMINATE_PENDING
]:
2584 new_state
= NetworkServiceRecordState
.VL_TERMINATE
2587 # If new state is RUNNING; check VNFFGRs are also active
2588 if new_state
== NetworkServiceRecordState
.RUNNING
:
2589 for _
, vnffgr
in self
.vnffgrs
.items():
2590 self
._log
.info("Checking vnffgr state for nsr %s is: %s",
2591 self
.id, vnffgr
.state
)
2592 if vnffgr
.state
== VnffgRecordState
.ACTIVE
:
2594 elif vnffgr
.state
== VnffgRecordState
.FAILED
:
2595 event_descr
= "Instantiation of VNFFGR %s failed" % vnffgr
.id
2596 self
.record_event("vnffg-failed", event_descr
)
2597 new_state
= NetworkServiceRecordState
.FAILED
2600 self
._log
.info("VNFFGR %s in NSR %s is still not active; current state is: %s",
2601 vnffgr
.id, self
.id, vnffgr
.state
)
2602 new_state
= curr_state
2604 # Update all the scaling group instance operational status to
2605 # reflect the state of all VNFR within that instance
2606 yield from self
._update
_scale
_group
_instances
_status
()
2608 for _
, group
in self
._scaling
_groups
.items():
2609 if group
.state
== scale_group
.ScaleGroupState
.SCALING_OUT
:
2610 new_state
= NetworkServiceRecordState
.SCALING_OUT
2612 elif group
.state
== scale_group
.ScaleGroupState
.SCALING_IN
:
2613 new_state
= NetworkServiceRecordState
.SCALING_IN
2616 if new_state
!= curr_state
:
2617 self
._log
.debug("Changing state of Network service %s from %s to %s",
2618 self
.id, curr_state
, new_state
)
2619 if new_state
== NetworkServiceRecordState
.RUNNING
:
2620 yield from self
.is_active()
2621 elif new_state
== NetworkServiceRecordState
.FAILED
:
2622 # If the NS is already active and we entered scaling_in, scaling_out,
2623 # do not mark the NS as failing if scaling operation failed.
2624 if curr_state
in [NetworkServiceRecordState
.SCALING_OUT
,
2625 NetworkServiceRecordState
.SCALING_IN
] and self
._is
_active
:
2626 new_state
= NetworkServiceRecordState
.RUNNING
2627 self
.set_state(new_state
)
2629 yield from self
.instantiation_failed()
2631 self
.set_state(new_state
)
2633 yield from self
.publish()
2636 class InputParameterSubstitution(object):
2638 This class is responsible for substituting input parameters into an NSD.
2641 def __init__(self
, log
):
2642 """Create an instance of InputParameterSubstitution
2645 log - a logger for this object to use
2650 def __call__(self
, nsd
, nsr_config
):
2651 """Substitutes input parameters from the NSR config into the NSD
2653 This call modifies the provided NSD with the input parameters that are
2654 contained in the NSR config.
2657 nsd - a GI NSD object
2658 nsr_config - a GI NSR config object
2661 if nsd
is None or nsr_config
is None:
2664 # Create a lookup of the xpath elements that this descriptor allows
2666 optional_input_parameters
= set()
2667 for input_parameter
in nsd
.input_parameter_xpath
:
2668 optional_input_parameters
.add(input_parameter
.xpath
)
2670 # Apply the input parameters to the descriptor
2671 if nsr_config
.input_parameter
:
2672 for param
in nsr_config
.input_parameter
:
2673 if param
.xpath
not in optional_input_parameters
:
2674 msg
= "tried to set an invalid input parameter ({})"
2675 self
.log
.error(msg
.format(param
.xpath
))
2679 "input-parameter:{} = {}".format(
2686 xpath
.setxattr(nsd
, param
.xpath
, param
.value
)
2688 except Exception as e
:
2689 self
.log
.exception(e
)
2692 class NetworkServiceDescriptor(object):
2694 Network service descriptor class
2697 def __init__(self
, dts
, log
, loop
, nsd
, nsm
):
2709 """ Returns nsd id """
2714 """ Returns name of nsd """
2715 return self
._nsd
.name
2718 def ref_count(self
):
2719 """ Returns reference count"""
2720 return self
._ref
_count
2723 """ Returns whether nsd is in use or not """
2724 return True if self
.ref_count
> 0 else False
2727 """ Take a reference on this object """
2728 self
._ref
_count
+= 1
2731 """ Release reference on this object """
2732 if self
.ref_count
< 1:
2733 msg
= ("Unref on a NSD object - nsd id %s, ref_count = %s" %
2734 (self
.id, self
.ref_count
))
2735 self
._log
.critical(msg
)
2736 raise NetworkServiceDescriptorError(msg
)
2737 self
._ref
_count
-= 1
2741 """ Return the message associated with this NetworkServiceDescriptor"""
2745 def path_for_id(nsd_id
):
2746 """ Return path for the passed nsd_id"""
2747 return "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}'".format(nsd_id
)
2750 """ Return the message associated with this NetworkServiceDescriptor"""
2751 return NetworkServiceDescriptor
.path_for_id(self
.id)
2753 def update(self
, nsd
):
2754 """ Update the NSD descriptor """
2758 class NsdDtsHandler(object):
2759 """ The network service descriptor DTS handler """
2760 XPATH
= "C,/nsd:nsd-catalog/nsd:nsd"
2762 def __init__(self
, dts
, log
, loop
, nsm
):
2772 """ Return registration handle """
2777 """ Register for Nsd create/update/delete/read requests from dts """
2779 def on_apply(dts
, acg
, xact
, action
, scratch
):
2780 """Apply the configuration"""
2781 is_recovery
= xact
.xact
is None and action
== rwdts
.AppconfAction
.INSTALL
2782 self
._log
.debug("Got nsd apply cfg (xact:%s) (action:%s)",
2784 # Create/Update an NSD record
2785 for cfg
in self
._regh
.get_xact_elements(xact
):
2786 # Only interested in those NSD cfgs whose ID was received in prepare callback
2787 if cfg
.id in scratch
.get('nsds', []) or is_recovery
:
2788 self
._nsm
.update_nsd(cfg
)
2790 scratch
.pop('nsds', None)
2792 return RwTypes
.RwStatus
.SUCCESS
2795 def delete_nsd_libs(nsd_id
):
2796 """ Remove any files uploaded with NSD and stored under $RIFT_ARTIFACTS/libs/<id> """
2798 rift_artifacts_dir
= os
.environ
['RIFT_ARTIFACTS']
2799 nsd_dir
= os
.path
.join(rift_artifacts_dir
, 'launchpad/libs', nsd_id
)
2801 if os
.path
.exists (nsd_dir
):
2802 shutil
.rmtree(nsd_dir
, ignore_errors
=True)
2803 except Exception as e
:
2804 self
._log
.error("Exception in cleaning up NSD libs {}: {}".
2806 self
._log
.excpetion(e
)
2809 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2810 """ Prepare callback from DTS for NSD config """
2812 self
._log
.info("Got nsd prepare - config received nsd id %s, msg %s",
2815 fref
= ProtobufC
.FieldReference
.alloc()
2816 fref
.goto_whole_message(msg
.to_pbcm())
2818 if fref
.is_field_deleted():
2819 # Delete an NSD record
2820 self
._log
.debug("Deleting NSD with id %s", msg
.id)
2821 if self
._nsm
.nsd_in_use(msg
.id):
2822 self
._log
.debug("Cannot delete NSD in use - %s", msg
.id)
2823 err
= "Cannot delete an NSD in use - %s" % msg
.id
2824 raise NetworkServiceDescriptorRefCountExists(err
)
2826 yield from delete_nsd_libs(msg
.id)
2827 self
._nsm
.delete_nsd(msg
.id)
2829 # Add this NSD to scratch to create/update in apply callback
2830 nsds
= scratch
.setdefault('nsds', [])
2832 # acg._scratch['nsds'].append(msg.id)
2834 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2837 "Registering for NSD config using xpath: %s",
2838 NsdDtsHandler
.XPATH
,
2841 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2842 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2843 # Need a list in scratch to store NSDs to create/update later
2844 # acg._scratch['nsds'] = list()
2845 self
._regh
= acg
.register(
2846 xpath
=NsdDtsHandler
.XPATH
,
2847 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
2848 on_prepare
=on_prepare
)
2851 class VnfdDtsHandler(object):
2852 """ DTS handler for VNFD config changes """
2853 XPATH
= "C,/vnfd:vnfd-catalog/vnfd:vnfd"
2855 def __init__(self
, dts
, log
, loop
, nsm
):
2864 """ DTS registration handle """
2869 """ Register for VNFD configuration"""
2872 def on_apply(dts
, acg
, xact
, action
, scratch
):
2873 """Apply the configuration"""
2874 self
._log
.debug("Got NSM VNFD apply (xact: %s) (action: %s)(scr: %s)",
2875 xact
, action
, scratch
)
2877 # Create/Update a VNFD record
2878 for cfg
in self
._regh
.get_xact_elements(xact
):
2879 # Only interested in those VNFD cfgs whose ID was received in prepare callback
2880 if cfg
.id in scratch
.get('vnfds', []):
2881 self
._nsm
.update_vnfd(cfg
)
2883 for cfg
in self
._regh
.elements
:
2884 if cfg
.id in scratch
.get('deleted_vnfds', []):
2885 yield from self
._nsm
.delete_vnfd(cfg
.id)
2887 scratch
.pop('vnfds', None)
2888 scratch
.pop('deleted_vnfds', None)
2891 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2892 """ on prepare callback """
2893 self
._log
.debug("Got on prepare for VNFD (path: %s) (action: %s) (msg: %s)",
2894 ks_path
.to_xpath(RwNsmYang
.get_schema()), xact_info
.query_action
, msg
)
2896 fref
= ProtobufC
.FieldReference
.alloc()
2897 fref
.goto_whole_message(msg
.to_pbcm())
2899 # Handle deletes in prepare_callback, but adds/updates in apply_callback
2900 if fref
.is_field_deleted():
2901 self
._log
.debug("Adding msg to deleted field")
2902 deleted_vnfds
= scratch
.setdefault('deleted_vnfds', [])
2903 deleted_vnfds
.append(msg
.id)
2905 # Add this VNFD to scratch to create/update in apply callback
2906 vnfds
= scratch
.setdefault('vnfds', [])
2907 vnfds
.append(msg
.id)
2909 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2912 "Registering for VNFD config using xpath: %s",
2913 VnfdDtsHandler
.XPATH
,
2915 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2916 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2917 # Need a list in scratch to store VNFDs to create/update later
2918 # acg._scratch['vnfds'] = list()
2919 # acg._scratch['deleted_vnfds'] = list()
2920 self
._regh
= acg
.register(
2921 xpath
=VnfdDtsHandler
.XPATH
,
2922 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY
,
2923 on_prepare
=on_prepare
)
2925 class NsrRpcDtsHandler(object):
2926 """ The network service instantiation RPC DTS handler """
2927 EXEC_NSR_CONF_XPATH
= "I,/nsr:start-network-service"
2928 EXEC_NSR_CONF_O_XPATH
= "O,/nsr:start-network-service"
2929 NETCONF_IP_ADDRESS
= "127.0.0.1"
2931 RESTCONF_PORT
= 8888
2932 NETCONF_USER
= "admin"
2933 NETCONF_PW
= "admin"
2934 REST_BASE_V2_URL
= 'https://{}:{}/v2/api/'.format("127.0.0.1",8888)
2936 def __init__(self
, dts
, log
, loop
, nsm
):
2943 self
._ns
_regh
= None
2945 self
._manager
= None
2946 self
._nsr
_config
_url
= NsrRpcDtsHandler
.REST_BASE_V2_URL
+ 'config/ns-instance-config'
2948 self
._model
= RwYang
.Model
.create_libncx()
2949 self
._model
.load_schema_ypbc(RwNsrYang
.get_schema())
2953 """ Return the NS manager instance """
2957 def wrap_netconf_config_xml(xml
):
2958 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
2962 def _connect(self
, timeout_secs
=240):
2964 start_time
= time
.time()
2965 while (time
.time() - start_time
) < timeout_secs
:
2968 self
._log
.debug("Attemping NsmTasklet netconf connection.")
2970 manager
= yield from ncclient
.asyncio_manager
.asyncio_connect(
2972 host
=NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
,
2973 port
=NsrRpcDtsHandler
.NETCONF_PORT
,
2974 username
=NsrRpcDtsHandler
.NETCONF_USER
,
2975 password
=NsrRpcDtsHandler
.NETCONF_PW
,
2977 look_for_keys
=False,
2978 hostkey_verify
=False,
2983 except ncclient
.transport
.errors
.SSHError
as e
:
2984 self
._log
.warning("Netconf connection to launchpad %s failed: %s",
2985 NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
, str(e
))
2987 yield from asyncio
.sleep(5, loop
=self
._loop
)
2989 raise NsrInstantiationFailed("Failed to connect to Launchpad within %s seconds" %
2992 def _apply_ns_instance_config(self
,payload_dict
):
2993 #self._log.debug("At apply NS instance config with payload %s",payload_dict)
2994 req_hdr
= {'accept':'application/vnd.yang.data+json','content-type':'application/vnd.yang.data+json'}
2995 response
=requests
.post(self
._nsr
_config
_url
, headers
=req_hdr
, auth
=('admin', 'admin'),data
=payload_dict
,verify
=False)
3000 """ Register for NS monitoring read from dts """
3002 def on_ns_config_prepare(xact_info
, action
, ks_path
, msg
):
3003 """ prepare callback from dts start-network-service"""
3004 assert action
== rwdts
.QueryAction
.RPC
3006 rpc_op
= NsrYang
.YangOutput_Nsr_StartNetworkService
.from_dict({
3007 "nsr_id":str(uuid
.uuid4())
3010 if not ('name' in rpc_ip
and 'nsd_ref' in rpc_ip
and ('cloud_account' in rpc_ip
or 'om_datacenter' in rpc_ip
)):
3011 self
._log
.error("Mandatory parameters name or nsd_ref or cloud account not found in start-network-service {}".format(rpc_ip
))
3014 self
._log
.debug("start-network-service RPC input: {}".format(rpc_ip
))
3017 # Add used value to the pool
3018 self
._log
.debug("RPC output: {}".format(rpc_op
))
3020 nsd_copy
= self
.nsm
.get_nsd(rpc_ip
.nsd_ref
)
3022 #if not self._manager:
3023 # self._manager = yield from self._connect()
3025 self
._log
.debug("Configuring ns-instance-config with name %s nsd-ref: %s",
3026 rpc_ip
.name
, rpc_ip
.nsd_ref
)
3028 ns_instance_config_dict
= {"id":rpc_op
.nsr_id
, "admin_status":"ENABLED"}
3029 ns_instance_config_copy_dict
= {k
:v
for k
, v
in rpc_ip
.as_dict().items()
3030 if k
in RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr().fields
}
3031 ns_instance_config_dict
.update(ns_instance_config_copy_dict
)
3033 ns_instance_config
= RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.from_dict(ns_instance_config_dict
)
3034 ns_instance_config
.nsd
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_Nsd()
3035 ns_instance_config
.nsd
.from_dict(nsd_copy
.msg
.as_dict())
3037 payload_dict
= ns_instance_config
.to_json(self
._model
)
3038 #xml = ns_instance_config.to_xml_v2(self._model)
3039 #netconf_xml = self.wrap_netconf_config_xml(xml)
3041 #self._log.debug("Sending configure ns-instance-config xml to %s: %s",
3042 # netconf_xml, NsrRpcDtsHandler.NETCONF_IP_ADDRESS)
3043 self
._log
.debug("Sending configure ns-instance-config json to %s: %s",
3044 self
._nsr
_config
_url
,ns_instance_config
)
3046 #response = yield from self._manager.edit_config(
3048 # config=netconf_xml,
3050 response
= yield from self
._loop
.run_in_executor(
3052 self
._apply
_ns
_instance
_config
,
3055 response
.raise_for_status()
3056 self
._log
.debug("Received edit config response: %s", response
.json())
3058 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
,
3059 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
,
3061 except Exception as e
:
3062 self
._log
.error("Exception processing the "
3063 "start-network-service: {}".format(e
))
3064 self
._log
.exception(e
)
3065 xact_info
.respond_xpath(rwdts
.XactRspCode
.NACK
,
3066 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
)
3069 hdl_ns
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_ns_config_prepare
,)
3071 with self
._dts
.group_create() as group
:
3072 self
._ns
_regh
= group
.register(xpath
=NsrRpcDtsHandler
.EXEC_NSR_CONF_XPATH
,
3074 flags
=rwdts
.Flag
.PUBLISHER
,
3078 class NsrDtsHandler(object):
3079 """ The network service DTS handler """
3080 NSR_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr"
3081 SCALE_INSTANCE_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr/nsr:scaling-group/nsr:instance"
3082 KEY_PAIR_XPATH
= "C,/nsr:key-pair"
3084 def __init__(self
, dts
, log
, loop
, nsm
):
3090 self
._nsr
_regh
= None
3091 self
._scale
_regh
= None
3092 self
._key
_pair
_regh
= None
3096 """ Return the NS manager instance """
3101 """ Register for Nsr create/update/delete/read requests from dts """
3103 def nsr_id_from_keyspec(ks
):
3104 nsr_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.schema().keyspec_to_entry(ks
)
3105 nsr_id
= nsr_path_entry
.key00
.id
3108 def group_name_from_keyspec(ks
):
3109 group_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
.schema().keyspec_to_entry(ks
)
3110 group_name
= group_path_entry
.key00
.scaling_group_name_ref
3113 def is_instance_in_reg_elements(nsr_id
, group_name
, instance_id
):
3114 """ Return boolean indicating if scaling group instance was already commited previously.
3116 By looking at the existing elements in this registration handle (elements not part
3117 of this current xact), we can tell if the instance was configured previously without
3118 keeping any application state.
3120 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3121 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3122 elem_group_name
= group_name_from_keyspec(keyspec
)
3124 if elem_nsr_id
!= nsr_id
or group_name
!= elem_group_name
:
3127 if instance_cfg
.id == instance_id
:
3132 def get_scale_group_instance_delta(nsr_id
, group_name
, xact
):
3133 delta
= {"added": [], "deleted": []}
3134 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3135 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3136 if elem_nsr_id
!= nsr_id
:
3139 elem_group_name
= group_name_from_keyspec(keyspec
)
3140 if elem_group_name
!= group_name
:
3143 delta
["added"].append(instance_cfg
.id)
3145 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(include_keyspec
=True):
3146 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3147 if elem_nsr_id
!= nsr_id
:
3150 elem_group_name
= group_name_from_keyspec(keyspec
)
3151 if elem_group_name
!= group_name
:
3154 if instance_cfg
.id in delta
["added"]:
3155 delta
["added"].remove(instance_cfg
.id)
3157 delta
["deleted"].append(instance_cfg
.id)
3162 def update_nsr_nsd(nsr_id
, xact
, scratch
):
3165 def get_nsr_vl_delta(nsr_id
, xact
, scratch
):
3166 delta
= {"added": [], "deleted": []}
3167 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3168 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3169 if elem_nsr_id
!= nsr_id
:
3172 if 'vld' in instance_cfg
.nsd
:
3173 for vld
in instance_cfg
.nsd
.vld
:
3174 delta
["added"].append(vld
)
3176 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3177 self
._log
.debug("NSR update: %s", instance_cfg
)
3178 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3179 if elem_nsr_id
!= nsr_id
:
3182 if 'vld' in instance_cfg
.nsd
:
3183 for vld
in instance_cfg
.nsd
.vld
:
3184 if vld
in delta
["added"]:
3185 delta
["added"].remove(vld
)
3187 delta
["deleted"].append(vld
)
3191 vl_delta
= yield from get_nsr_vl_delta(nsr_id
, xact
, scratch
)
3192 self
._log
.debug("Got NSR:%s VL instance delta: %s", nsr_id
, vl_delta
)
3194 for vld
in vl_delta
["added"]:
3195 yield from self
._nsm
.nsr_instantiate_vl(nsr_id
, vld
)
3197 for vld
in vl_delta
["deleted"]:
3198 yield from self
._nsm
.nsr_terminate_vl(nsr_id
, vld
)
3200 def get_add_delete_update_cfgs(dts_member_reg
, xact
, key_name
, scratch
):
3201 # Unfortunately, it is currently difficult to figure out what has exactly
3202 # changed in this xact without Pbdelta support (RIFT-4916)
3203 # As a workaround, we can fetch the pre and post xact elements and
3204 # perform a comparison to figure out adds/deletes/updates
3205 xact_cfgs
= list(dts_member_reg
.get_xact_elements(xact
))
3206 curr_cfgs
= list(dts_member_reg
.elements
)
3208 xact_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in xact_cfgs
}
3209 curr_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in curr_cfgs
}
3212 added_keys
= set(xact_key_map
) - set(curr_key_map
)
3213 added_cfgs
= [xact_key_map
[key
] for key
in added_keys
]
3216 deleted_keys
= set(curr_key_map
) - set(xact_key_map
)
3217 deleted_cfgs
= [curr_key_map
[key
] for key
in deleted_keys
]
3220 updated_keys
= set(curr_key_map
) & set(xact_key_map
)
3221 updated_cfgs
= [xact_key_map
[key
] for key
in updated_keys
3222 if xact_key_map
[key
] != curr_key_map
[key
]]
3224 return added_cfgs
, deleted_cfgs
, updated_cfgs
3226 def get_nsr_key_pairs(dts_member_reg
, xact
):
3228 for instance_cfg
, keyspec
in dts_member_reg
.get_xact_elements(xact
, include_keyspec
=True):
3229 self
._log
.debug("Key pair received is {} KS: {}".format(instance_cfg
, keyspec
))
3230 xpath
= keyspec
.to_xpath(RwNsrYang
.get_schema())
3231 key_pairs
[instance_cfg
.name
] = instance_cfg
3234 def on_apply(dts
, acg
, xact
, action
, scratch
):
3235 """Apply the configuration"""
3236 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3237 xact
, action
, scratch
)
3239 def handle_create_nsr(msg
, key_pairs
=None, restart_mode
=False):
3240 # Handle create nsr requests """
3241 # Do some validations
3242 if not msg
.has_field("nsd"):
3243 err
= "NSD not provided"
3244 self
._log
.error(err
)
3245 raise NetworkServiceRecordError(err
)
3247 self
._log
.debug("Creating NetworkServiceRecord %s from nsr config %s",
3248 msg
.id, msg
.as_dict())
3249 nsr
= self
.nsm
.create_nsr(msg
, key_pairs
=key_pairs
, restart_mode
=restart_mode
)
3252 def handle_delete_nsr(msg
):
3254 def delete_instantiation(ns_id
):
3255 """ Delete instantiation """
3256 with self
._dts
.transaction() as xact
:
3257 yield from self
._nsm
.terminate_ns(ns_id
, xact
)
3259 # Handle delete NSR requests
3260 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3261 # Terminate the NSR instance
3262 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3264 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3265 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3266 nsr
.record_event("terminate-rcvd", event_descr
)
3268 self
._loop
.create_task(delete_instantiation(msg
.id))
3271 def begin_instantiation(nsr
):
3272 # Begin instantiation
3273 self
._log
.info("Beginning NS instantiation: %s", nsr
.id)
3274 yield from self
._nsm
.instantiate_ns(nsr
.id, xact
)
3276 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3277 xact
, action
, scratch
)
3279 if action
== rwdts
.AppconfAction
.INSTALL
and xact
.id is None:
3281 for element
in self
._key
_pair
_regh
.elements
:
3282 key_pairs
.append(element
)
3283 for element
in self
._nsr
_regh
.elements
:
3284 nsr
= handle_create_nsr(element
, key_pairs
, restart_mode
=True)
3285 self
._loop
.create_task(begin_instantiation(nsr
))
3288 (added_msgs
, deleted_msgs
, updated_msgs
) = get_add_delete_update_cfgs(self
._nsr
_regh
,
3292 self
._log
.debug("Added: %s, Deleted: %s, Updated: %s", added_msgs
,
3293 deleted_msgs
, updated_msgs
)
3295 for msg
in added_msgs
:
3296 if msg
.id not in self
._nsm
.nsrs
:
3297 self
._log
.info("Create NSR received in on_apply to instantiate NS:%s", msg
.id)
3298 key_pairs
= get_nsr_key_pairs(self
._key
_pair
_regh
, xact
)
3299 nsr
= handle_create_nsr(msg
,key_pairs
)
3300 self
._loop
.create_task(begin_instantiation(nsr
))
3302 for msg
in deleted_msgs
:
3303 self
._log
.info("Delete NSR received in on_apply to terminate NS:%s", msg
.id)
3305 handle_delete_nsr(msg
)
3307 self
._log
.exception("Failed to terminate NS:%s", msg
.id)
3309 for msg
in updated_msgs
:
3310 self
._log
.info("Update NSR received in on_apply: %s", msg
)
3312 self
._nsm
.nsr_update_cfg(msg
.id, msg
)
3315 self
._loop
.create_task(update_nsr_nsd(msg
.id, xact
, scratch
))
3317 for group
in msg
.scaling_group
:
3318 instance_delta
= get_scale_group_instance_delta(msg
.id, group
.scaling_group_name_ref
, xact
)
3319 self
._log
.debug("Got NSR:%s scale group instance delta: %s", msg
.id, instance_delta
)
3321 for instance_id
in instance_delta
["added"]:
3322 self
._nsm
.scale_nsr_out(msg
.id, group
.scaling_group_name_ref
, instance_id
, xact
)
3324 for instance_id
in instance_delta
["deleted"]:
3325 self
._nsm
.scale_nsr_in(msg
.id, group
.scaling_group_name_ref
, instance_id
)
3328 return RwTypes
.RwStatus
.SUCCESS
3331 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
3332 """ Prepare calllback from DTS for NSR """
3334 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3335 action
= xact_info
.query_action
3337 "Got Nsr prepare callback (xact: %s) (action: %s) (info: %s), %s:%s)",
3338 xact
, action
, xact_info
, xpath
, msg
3342 def delete_instantiation(ns_id
):
3343 """ Delete instantiation """
3344 yield from self
._nsm
.terminate_ns(ns_id
, None)
3346 def handle_delete_nsr():
3347 """ Handle delete NSR requests """
3348 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3349 # Terminate the NSR instance
3350 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3352 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3353 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3354 nsr
.record_event("terminate-rcvd", event_descr
)
3356 self
._loop
.create_task(delete_instantiation(msg
.id))
3358 fref
= ProtobufC
.FieldReference
.alloc()
3359 fref
.goto_whole_message(msg
.to_pbcm())
3361 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
, rwdts
.QueryAction
.DELETE
]:
3362 # if this is an NSR create
3363 if action
!= rwdts
.QueryAction
.DELETE
and msg
.id not in self
._nsm
.nsrs
:
3364 # Ensure the Cloud account/datacenter has been specified
3365 if not msg
.has_field("cloud_account") and not msg
.has_field("om_datacenter"):
3366 raise NsrInstantiationFailed("Cloud account or datacenter not specified in NSR")
3368 # Check if nsd is specified
3369 if not msg
.has_field("nsd"):
3370 raise NsrInstantiationFailed("NSD not specified in NSR")
3373 nsr
= self
._nsm
.nsrs
[msg
.id]
3375 if msg
.has_field("nsd"):
3376 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3377 raise NsrVlUpdateError("Unable to update VL when NSR not in running state")
3378 if 'vld' not in msg
.nsd
or len(msg
.nsd
.vld
) == 0:
3379 raise NsrVlUpdateError("NS config NSD should have atleast 1 VLD defined")
3381 if msg
.has_field("scaling_group"):
3382 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3383 raise ScalingOperationError("Unable to perform scaling action when NS is not in running state")
3385 if len(msg
.scaling_group
) > 1:
3386 raise ScalingOperationError("Only a single scaling group can be configured at a time")
3388 for group_msg
in msg
.scaling_group
:
3389 num_new_group_instances
= len(group_msg
.instance
)
3390 if num_new_group_instances
> 1:
3391 raise ScalingOperationError("Only a single scaling instance can be modified at a time")
3393 elif num_new_group_instances
== 1:
3394 scale_group
= nsr
.scaling_groups
[group_msg
.scaling_group_name_ref
]
3395 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
]:
3396 if len(scale_group
.instances
) == scale_group
.max_instance_count
:
3397 raise ScalingOperationError("Max instances for %s reached" % scale_group
)
3399 acg
.handle
.prepare_complete_ok(xact_info
.handle
)
3402 self
._log
.debug("Registering for NSR config using xpath: %s",
3403 NsrDtsHandler
.NSR_XPATH
)
3405 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
3406 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
3407 self
._nsr
_regh
= acg
.register(xpath
=NsrDtsHandler
.NSR_XPATH
,
3408 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3409 on_prepare
=on_prepare
)
3411 self
._scale
_regh
= acg
.register(
3412 xpath
=NsrDtsHandler
.SCALE_INSTANCE_XPATH
,
3413 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY| rwdts
.Flag
.CACHE
,
3416 self
._key
_pair
_regh
= acg
.register(
3417 xpath
=NsrDtsHandler
.KEY_PAIR_XPATH
,
3418 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3422 class NsrOpDataDtsHandler(object):
3423 """ The network service op data DTS handler """
3424 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
3426 def __init__(self
, dts
, log
, loop
, nsm
):
3435 """ Return the registration handle"""
3440 """ Return the NS manager instance """
3445 """ Register for Nsr op data publisher registration"""
3446 self
._log
.debug("Registering Nsr op data path %s as publisher",
3447 NsrOpDataDtsHandler
.XPATH
)
3449 hdl
= rift
.tasklets
.DTS
.RegistrationHandler()
3450 handlers
= rift
.tasklets
.Group
.Handler()
3451 with self
._dts
.group_create(handler
=handlers
) as group
:
3452 self
._regh
= group
.register(xpath
=NsrOpDataDtsHandler
.XPATH
,
3454 flags
=rwdts
.Flag
.PUBLISHER | rwdts
.Flag
.NO_PREP_READ | rwdts
.Flag
.DATASTORE
)
3457 def create(self
, path
, msg
):
3459 Create an NS record in DTS with the path and message
3461 self
._log
.debug("Creating NSR %s:%s", path
, msg
)
3462 self
.regh
.create_element(path
, msg
)
3463 self
._log
.debug("Created NSR, %s:%s", path
, msg
)
3466 def update(self
, path
, msg
, flags
=rwdts
.XactFlag
.REPLACE
):
3468 Update an NS record in DTS with the path and message
3470 self
._log
.debug("Updating NSR, %s:%s regh = %s", path
, msg
, self
.regh
)
3471 self
.regh
.update_element(path
, msg
, flags
)
3472 self
._log
.debug("Updated NSR, %s:%s", path
, msg
)
3475 def delete(self
, path
):
3477 Update an NS record in DTS with the path and message
3479 self
._log
.debug("Deleting NSR path:%s", path
)
3480 self
.regh
.delete_element(path
)
3481 self
._log
.debug("Deleted NSR path:%s", path
)
3484 class VnfrDtsHandler(object):
3485 """ The virtual network service DTS handler """
3486 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
3488 def __init__(self
, dts
, log
, loop
, nsm
):
3498 """ Return registration handle """
3503 """ Return the NS manager instance """
3508 """ Register for vnfr create/update/delete/ advises from dts """
3510 def on_commit(xact_info
):
3511 """ The transaction has been committed """
3512 self
._log
.debug("Got vnfr commit (xact_info: %s)", xact_info
)
3513 return rwdts
.MemberRspCode
.ACTION_OK
3516 def on_prepare(xact_info
, action
, ks_path
, msg
):
3517 """ prepare callback from dts """
3518 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3520 "Got vnfr on_prepare cb (xact_info: %s, action: %s): %s:%s",
3521 xact_info
, action
, ks_path
, msg
3524 schema
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr
.schema()
3525 path_entry
= schema
.keyspec_to_entry(ks_path
)
3526 if path_entry
.key00
.id not in self
._nsm
._vnfrs
:
3527 self
._log
.error("%s request for non existent record path %s",
3529 xact_info
.respond_xpath(rwdts
.XactRspCode
.NA
, xpath
)
3533 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3534 if action
== rwdts
.QueryAction
.CREATE
or action
== rwdts
.QueryAction
.UPDATE
:
3535 yield from self
._nsm
.update_vnfr(msg
)
3536 elif action
== rwdts
.QueryAction
.DELETE
:
3537 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3538 self
._nsm
.delete_vnfr(path_entry
.key00
.id)
3540 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
, xpath
)
3542 self
._log
.debug("Registering for VNFR using xpath: %s",
3543 VnfrDtsHandler
.XPATH
,)
3545 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_commit
=on_commit
,
3546 on_prepare
=on_prepare
,)
3547 with self
._dts
.group_create() as group
:
3548 self
._regh
= group
.register(xpath
=VnfrDtsHandler
.XPATH
,
3550 flags
=(rwdts
.Flag
.SUBSCRIBER
),)
3553 class NsdRefCountDtsHandler(object):
3554 """ The NSD Ref Count DTS handler """
3555 XPATH
= "D,/nsr:ns-instance-opdata/rw-nsr:nsd-ref-count"
3557 def __init__(self
, dts
, log
, loop
, nsm
):
3567 """ Return registration handle """
3572 """ Return the NS manager instance """
3577 """ Register for NSD ref count read from dts """
3580 def on_prepare(xact_info
, action
, ks_path
, msg
):
3581 """ prepare callback from dts """
3582 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3584 if action
== rwdts
.QueryAction
.READ
:
3585 schema
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount
.schema()
3586 path_entry
= schema
.keyspec_to_entry(ks_path
)
3587 nsd_list
= yield from self
._nsm
.get_nsd_refcount(path_entry
.key00
.nsd_id_ref
)
3588 for xpath
, msg
in nsd_list
:
3589 xact_info
.respond_xpath(rsp_code
=rwdts
.XactRspCode
.MORE
,
3592 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
3594 raise NetworkServiceRecordError("Not supported operation %s" % action
)
3596 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_prepare
,)
3597 with self
._dts
.group_create() as group
:
3598 self
._regh
= group
.register(xpath
=NsdRefCountDtsHandler
.XPATH
,
3600 flags
=rwdts
.Flag
.PUBLISHER
,)
3603 class NsManager(object):
3604 """ The Network Service Manager class"""
3605 def __init__(self
, dts
, log
, loop
,
3606 nsr_handler
, vnfr_handler
, vlr_handler
, ro_plugin_selector
,
3607 vnffgmgr
, vnfd_pub_handler
, cloud_account_handler
):
3611 self
._nsr
_handler
= nsr_handler
3612 self
._vnfr
_pub
_handler
= vnfr_handler
3613 self
._vlr
_pub
_handler
= vlr_handler
3614 self
._vnffgmgr
= vnffgmgr
3615 self
._vnfd
_pub
_handler
= vnfd_pub_handler
3616 self
._cloud
_account
_handler
= cloud_account_handler
3618 self
._ro
_plugin
_selector
= ro_plugin_selector
3619 self
._ncclient
= rift
.mano
.ncclient
.NcClient(
3631 self
.cfgmgr_obj
= conman
.ROConfigManager(log
, loop
, dts
, self
)
3633 # TODO: All these handlers should move to tasklet level.
3634 # Passing self is often an indication of bad design
3635 self
._nsd
_dts
_handler
= NsdDtsHandler(dts
, log
, loop
, self
)
3636 self
._vnfd
_dts
_handler
= VnfdDtsHandler(dts
, log
, loop
, self
)
3637 self
._dts
_handlers
= [self
._nsd
_dts
_handler
,
3638 VnfrDtsHandler(dts
, log
, loop
, self
),
3639 NsdRefCountDtsHandler(dts
, log
, loop
, self
),
3640 NsrDtsHandler(dts
, log
, loop
, self
),
3641 ScalingRpcHandler(log
, dts
, loop
, self
.scale_rpc_callback
),
3642 NsrRpcDtsHandler(dts
,log
,loop
,self
),
3643 self
._vnfd
_dts
_handler
,
3664 def nsr_handler(self
):
3665 """" NSR handler """
3666 return self
._nsr
_handler
3670 """" So Obj handler """
3675 """ NSRs in this NSM"""
3680 """ NSDs in this NSM"""
3685 """ VNFDs in this NSM"""
3690 """ VNFRs in this NSM"""
3694 def nsr_pub_handler(self
):
3695 """ NSR publication handler """
3696 return self
._nsr
_handler
3699 def vnfr_pub_handler(self
):
3700 """ VNFR publication handler """
3701 return self
._vnfr
_pub
_handler
3704 def vlr_pub_handler(self
):
3705 """ VLR publication handler """
3706 return self
._vlr
_pub
_handler
3709 def vnfd_pub_handler(self
):
3710 return self
._vnfd
_pub
_handler
3714 """ Register all static DTS handlers """
3715 for dts_handle
in self
._dts
_handlers
:
3716 yield from dts_handle
.register()
3719 def get_ns_by_nsr_id(self
, nsr_id
):
3720 """ get NSR by nsr id """
3721 if nsr_id
not in self
._nsrs
:
3722 raise NetworkServiceRecordError("NSR id %s not found" % nsr_id
)
3724 return self
._nsrs
[nsr_id
]
3726 def scale_nsr_out(self
, nsr_id
, scale_group_name
, instance_id
, config_xact
):
3727 self
.log
.debug("Scale out NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3732 nsr
= self
._nsrs
[nsr_id
]
3733 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3734 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3736 self
._loop
.create_task(nsr
.create_scale_group_instance(scale_group_name
, instance_id
, config_xact
))
3738 def scale_nsr_in(self
, nsr_id
, scale_group_name
, instance_id
):
3739 self
.log
.debug("Scale in NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3744 nsr
= self
._nsrs
[nsr_id
]
3745 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3746 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3748 self
._loop
.create_task(nsr
.delete_scale_group_instance(scale_group_name
, instance_id
))
3750 def scale_rpc_callback(self
, xact
, msg
, action
):
3751 """Callback handler for RPC calls
3753 xact : Transaction Handler
3755 action : Scaling Action
3757 ScalingGroupInstance
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup_Instance
3758 ScalingGroup
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
3760 xpath
= ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').format(
3762 instance
= ScalingGroupInstance
.from_dict({"id": msg
.instance_id
})
3765 def get_nsr_scaling_group():
3766 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3768 for result
in results
:
3769 res
= yield from result
3770 nsr_config
= res
.result
3772 for scaling_group
in nsr_config
.scaling_group
:
3773 if scaling_group
.scaling_group_name_ref
== msg
.scaling_group_name_ref
:
3776 scaling_group
= nsr_config
.scaling_group
.add()
3777 scaling_group
.scaling_group_name_ref
= msg
.scaling_group_name_ref
3779 return (nsr_config
, scaling_group
)
3782 def update_config(nsr_config
):
3783 xml
= self
._ncclient
.convert_to_xml(RwNsrYang
, nsr_config
)
3784 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
3785 yield from self
._ncclient
.connect()
3786 yield from self
._ncclient
.manager
.edit_config(target
="running", config
=xml
, default_operation
="replace")
3790 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3791 scaling_group
.instance
.append(instance
)
3792 yield from update_config(nsr_config
)
3796 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3797 scaling_group
.instance
.remove(instance
)
3798 yield from update_config(nsr_config
)
3800 if action
== ScalingRpcHandler
.ACTION
.SCALE_OUT
:
3801 self
._loop
.create_task(scale_out())
3803 self
._loop
.create_task(scale_in())
3805 # Opdata based calls, disabled for now!
3806 # if action == ScalingRpcHandler.ACTION.SCALE_OUT:
3807 # self.scale_nsr_out(
3809 # msg.scaling_group_name_ref,
3813 # self.scale_nsr_in(
3815 # msg.scaling_group_name_ref,
3818 def nsr_update_cfg(self
, nsr_id
, msg
):
3819 nsr
= self
._nsrs
[nsr_id
]
3820 nsr
.nsr_cfg_msg
= msg
3822 def nsr_instantiate_vl(self
, nsr_id
, vld
):
3823 self
.log
.debug("NSR {} create VL {}".format(nsr_id
, vld
))
3824 nsr
= self
._nsrs
[nsr_id
]
3825 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3826 raise NsrVlUpdateError("Cannot perform VL instantiate if NSR is not in running state")
3828 # Not calling in a separate task as this is called from a separate task
3829 yield from nsr
.create_vl_instance(vld
)
3831 def nsr_terminate_vl(self
, nsr_id
, vld
):
3832 self
.log
.debug("NSR {} delete VL {}".format(nsr_id
, vld
.id))
3833 nsr
= self
._nsrs
[nsr_id
]
3834 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3835 raise NsrVlUpdateError("Cannot perform VL terminate if NSR is not in running state")
3837 # Not calling in a separate task as this is called from a separate task
3838 yield from nsr
.delete_vl_instance(vld
)
3840 def create_nsr(self
, nsr_msg
, key_pairs
=None,restart_mode
=False):
3841 """ Create an NSR instance """
3842 if nsr_msg
.id in self
._nsrs
:
3843 msg
= "NSR id %s already exists" % nsr_msg
.id
3844 self
._log
.error(msg
)
3845 raise NetworkServiceRecordError(msg
)
3847 self
._log
.info("Create NetworkServiceRecord nsr id %s from nsd_id %s",
3851 nsm_plugin
= self
._ro
_plugin
_selector
.ro_plugin
3852 sdn_account_name
= self
._cloud
_account
_handler
.get_cloud_account_sdn_name(nsr_msg
.cloud_account
)
3854 nsr
= NetworkServiceRecord(self
._dts
,
3862 restart_mode
=restart_mode
,
3863 vlr_handler
=self
._ro
_plugin
_selector
._records
_publisher
._vlr
_pub
_hdlr
3865 self
._nsrs
[nsr_msg
.id] = nsr
3866 nsm_plugin
.create_nsr(nsr_msg
, nsr_msg
.nsd
, key_pairs
)
3870 def delete_nsr(self
, nsr_id
):
3872 Delete NSR with the passed nsr id
3874 del self
._nsrs
[nsr_id
]
3877 def instantiate_ns(self
, nsr_id
, config_xact
):
3878 """ Instantiate an NS instance """
3879 self
._log
.debug("Instantiating Network service id %s", nsr_id
)
3880 if nsr_id
not in self
._nsrs
:
3881 err
= "NSR id %s not found " % nsr_id
3882 self
._log
.error(err
)
3883 raise NetworkServiceRecordError(err
)
3885 nsr
= self
._nsrs
[nsr_id
]
3886 yield from nsr
.nsm_plugin
.instantiate_ns(nsr
, config_xact
)
3889 def update_vnfr(self
, vnfr
):
3890 """Create/Update an VNFR """
3892 vnfr_state
= self
._vnfrs
[vnfr
.id].state
3893 self
._log
.debug("Updating VNFR with state %s: vnfr %s", vnfr_state
, vnfr
)
3895 yield from self
._vnfrs
[vnfr
.id].update_state(vnfr
)
3896 nsr
= self
.find_nsr_for_vnfr(vnfr
.id)
3897 yield from nsr
.update_state()
3899 def find_nsr_for_vnfr(self
, vnfr_id
):
3900 """ Find the NSR which )has the passed vnfr id"""
3901 for nsr
in list(self
.nsrs
.values()):
3902 for vnfr
in list(nsr
.vnfrs
.values()):
3903 if vnfr
.id == vnfr_id
:
3907 def delete_vnfr(self
, vnfr_id
):
3908 """ Delete VNFR with the passed id"""
3909 del self
._vnfrs
[vnfr_id
]
3911 def get_nsd_ref(self
, nsd_id
):
3912 """ Get network service descriptor for the passed nsd_id
3914 nsd
= self
.get_nsd(nsd_id
)
3919 def get_nsr_config(self
, nsd_id
):
3920 xpath
= "C,/nsr:ns-instance-config"
3921 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3923 for result
in results
:
3924 entry
= yield from result
3925 ns_instance_config
= entry
.result
3927 for nsr
in ns_instance_config
.nsr
:
3928 if nsr
.nsd
.id == nsd_id
:
3934 def nsd_unref_by_nsr_id(self
, nsr_id
):
3935 """ Unref the network service descriptor based on NSR id """
3936 self
._log
.debug("NSR Unref called for Nsr Id:%s", nsr_id
)
3937 if nsr_id
in self
._nsrs
:
3938 nsr
= self
._nsrs
[nsr_id
]
3941 nsd
= self
.get_nsd(nsr
.nsd_id
)
3942 self
._log
.debug("Releasing ref on NSD %s held by NSR %s - Curr %d",
3943 nsd
.id, nsr
.id, nsd
.ref_count
)
3945 except NetworkServiceDescriptorError
:
3946 # We store a copy of NSD in NSR and the NSD in nsd-catalog
3951 self
._log
.error("Cannot find NSR with id %s", nsr_id
)
3952 raise NetworkServiceDescriptorUnrefError("No NSR with id" % nsr_id
)
3955 def nsd_unref(self
, nsd_id
):
3956 """ Unref the network service descriptor associated with the id """
3957 nsd
= self
.get_nsd(nsd_id
)
3960 def get_nsd(self
, nsd_id
):
3961 """ Get network service descriptor for the passed nsd_id"""
3962 if nsd_id
not in self
._nsds
:
3963 self
._log
.error("Cannot find NSD id:%s", nsd_id
)
3964 raise NetworkServiceDescriptorError("Cannot find NSD id:%s", nsd_id
)
3966 return self
._nsds
[nsd_id
]
3968 def create_nsd(self
, nsd_msg
):
3969 """ Create a network service descriptor """
3970 self
._log
.debug("Create network service descriptor - %s", nsd_msg
)
3971 if nsd_msg
.id in self
._nsds
:
3972 self
._log
.error("Cannot create NSD %s -NSD ID already exists", nsd_msg
)
3973 raise NetworkServiceDescriptorError("NSD already exists-%s", nsd_msg
.id)
3975 nsd
= NetworkServiceDescriptor(
3982 self
._nsds
[nsd_msg
.id] = nsd
3986 def update_nsd(self
, nsd
):
3987 """ update the Network service descriptor """
3988 self
._log
.debug("Update network service descriptor - %s", nsd
)
3989 if nsd
.id not in self
._nsds
:
3990 self
._log
.debug("No NSD found - creating NSD id = %s", nsd
.id)
3991 self
.create_nsd(nsd
)
3993 self
._log
.debug("Updating NSD id = %s, nsd = %s", nsd
.id, nsd
)
3994 self
._nsds
[nsd
.id].update(nsd
)
3996 def delete_nsd(self
, nsd_id
):
3997 """ Delete the Network service descriptor with the passed id """
3998 self
._log
.debug("Deleting the network service descriptor - %s", nsd_id
)
3999 if nsd_id
not in self
._nsds
:
4000 self
._log
.debug("Delete NSD failed - cannot find nsd-id %s", nsd_id
)
4001 raise NetworkServiceDescriptorNotFound("Cannot find %s", nsd_id
)
4003 if nsd_id
not in self
._nsds
:
4004 self
._log
.debug("Cannot delete NSD id %s reference exists %s",
4006 self
._nsds
[nsd_id
].ref_count
)
4007 raise NetworkServiceDescriptorRefCountExists(
4008 "Cannot delete :%s, ref_count:%s",
4010 self
._nsds
[nsd_id
].ref_count
)
4012 del self
._nsds
[nsd_id
]
4014 def get_vnfd_config(self
, xact
):
4015 vnfd_dts_reg
= self
._vnfd
_dts
_handler
.regh
4016 for cfg
in vnfd_dts_reg
.get_xact_elements(xact
):
4017 if cfg
.id not in self
._vnfds
:
4018 self
.create_vnfd(cfg
)
4020 def get_vnfd(self
, vnfd_id
, xact
):
4021 """ Get virtual network function descriptor for the passed vnfd_id"""
4022 if vnfd_id
not in self
._vnfds
:
4023 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4024 self
.get_vnfd_config(xact
)
4026 if vnfd_id
not in self
._vnfds
:
4027 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4028 raise VnfDescriptorError("Cannot find VNFD id:%s", vnfd_id
)
4030 return self
._vnfds
[vnfd_id
]
4032 def create_vnfd(self
, vnfd
):
4033 """ Create a virtual network function descriptor """
4034 self
._log
.debug("Create virtual network function descriptor - %s", vnfd
)
4035 if vnfd
.id in self
._vnfds
:
4036 self
._log
.error("Cannot create VNFD %s -VNFD ID already exists", vnfd
)
4037 raise VnfDescriptorError("VNFD already exists-%s", vnfd
.id)
4039 self
._vnfds
[vnfd
.id] = vnfd
4040 return self
._vnfds
[vnfd
.id]
4042 def update_vnfd(self
, vnfd
):
4043 """ Update the virtual network function descriptor """
4044 self
._log
.debug("Update virtual network function descriptor- %s", vnfd
)
4047 if vnfd
.id not in self
._vnfds
:
4048 self
._log
.debug("No VNFD found - creating VNFD id = %s", vnfd
.id)
4049 self
.create_vnfd(vnfd
)
4051 self
._log
.debug("Updating VNFD id = %s, vnfd = %s", vnfd
.id, vnfd
)
4052 self
._vnfds
[vnfd
.id] = vnfd
4055 def delete_vnfd(self
, vnfd_id
):
4056 """ Delete the virtual network function descriptor with the passed id """
4057 self
._log
.debug("Deleting the virtual network function descriptor - %s", vnfd_id
)
4058 if vnfd_id
not in self
._vnfds
:
4059 self
._log
.debug("Delete VNFD failed - cannot find vnfd-id %s", vnfd_id
)
4060 raise VnfDescriptorError("Cannot find %s", vnfd_id
)
4062 del self
._vnfds
[vnfd_id
]
4064 def nsd_in_use(self
, nsd_id
):
4065 """ Is the NSD with the passed id in use """
4066 self
._log
.debug("Is this NSD in use - msg:%s", nsd_id
)
4067 if nsd_id
in self
._nsds
:
4068 return self
._nsds
[nsd_id
].in_use()
4072 def publish_nsr(self
, xact
, path
, msg
):
4073 """ Publish a NSR """
4074 self
._log
.debug("Publish NSR with path %s, msg %s",
4076 yield from self
.nsr_handler
.update(xact
, path
, msg
)
4079 def unpublish_nsr(self
, xact
, path
):
4080 """ Un Publish an NSR """
4081 self
._log
.debug("Publishing delete NSR with path %s", path
)
4082 yield from self
.nsr_handler
.delete(path
, xact
)
4084 def vnfr_is_ready(self
, vnfr_id
):
4085 """ VNFR with the id is ready """
4086 self
._log
.debug("VNFR id %s ready", vnfr_id
)
4087 if vnfr_id
not in self
._vnfds
:
4088 err
= "Did not find VNFR ID with id %s" % vnfr_id
4089 self
._log
.critical("err")
4090 raise VirtualNetworkFunctionRecordError(err
)
4091 self
._vnfrs
[vnfr_id
].is_ready()
4094 def get_nsd_refcount(self
, nsd_id
):
4095 """ Get the nsd_list from this NSM"""
4097 def nsd_refcount_xpath(nsd_id
):
4098 """ xpath for ref count entry """
4099 return (NsdRefCountDtsHandler
.XPATH
+
4100 "[rw-nsr:nsd-id-ref = '{}']").format(nsd_id
)
4103 if nsd_id
is None or nsd_id
== "":
4104 for nsd
in self
._nsds
.values():
4105 nsd_msg
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4106 nsd_msg
.nsd_id_ref
= nsd
.id
4107 nsd_msg
.instance_ref_count
= nsd
.ref_count
4108 nsd_list
.append((nsd_refcount_xpath(nsd
.id), nsd_msg
))
4109 elif nsd_id
in self
._nsds
:
4110 nsd_msg
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4111 nsd_msg
.nsd_id_ref
= self
._nsds
[nsd_id
].id
4112 nsd_msg
.instance_ref_count
= self
._nsds
[nsd_id
].ref_count
4113 nsd_list
.append((nsd_refcount_xpath(nsd_id
), nsd_msg
))
4118 def terminate_ns(self
, nsr_id
, xact
):
4120 Terminate network service for the given NSR Id
4123 # Terminate the instances/networks assocaited with this nw service
4124 self
._log
.debug("Terminating the network service %s", nsr_id
)
4126 yield from self
._nsrs
[nsr_id
].terminate()
4127 except Exception as e
:
4128 self
.log
.exception("Failed to terminate NSR[id=%s]", nsr_id
)
4131 yield from self
.nsd_unref_by_nsr_id(nsr_id
)
4133 # Unpublish the NSR record
4134 self
._log
.debug("Unpublishing the network service %s", nsr_id
)
4135 yield from self
._nsrs
[nsr_id
].unpublish(xact
)
4137 # Finaly delete the NS instance from this NS Manager
4138 self
._log
.debug("Deletng the network service %s", nsr_id
)
4139 self
.delete_nsr(nsr_id
)
4142 class NsmRecordsPublisherProxy(object):
4143 """ This class provides a publisher interface that allows plugin objects
4144 to publish NSR/VNFR/VLR"""
4146 def __init__(self
, dts
, log
, loop
, nsr_pub_hdlr
, vnfr_pub_hdlr
, vlr_pub_hdlr
):
4150 self
._nsr
_pub
_hdlr
= nsr_pub_hdlr
4151 self
._vlr
_pub
_hdlr
= vlr_pub_hdlr
4152 self
._vnfr
_pub
_hdlr
= vnfr_pub_hdlr
4155 def publish_nsr(self
, xact
, nsr
):
4156 """ Publish an NSR """
4157 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4158 return (yield from self
._nsr
_pub
_hdlr
.update(xact
, path
, nsr
))
4161 def unpublish_nsr(self
, xact
, nsr
):
4162 """ Unpublish an NSR """
4163 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4164 return (yield from self
._nsr
_pub
_hdlr
.delete(xact
, path
))
4167 def publish_vnfr(self
, xact
, vnfr
):
4168 """ Publish an VNFR """
4169 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4170 return (yield from self
._vnfr
_pub
_hdlr
.update(xact
, path
, vnfr
))
4173 def unpublish_vnfr(self
, xact
, vnfr
):
4174 """ Unpublish a VNFR """
4175 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4176 return (yield from self
._vnfr
_pub
_hdlr
.delete(xact
, path
))
4179 def publish_vlr(self
, xact
, vlr
):
4180 """ Publish a VLR """
4181 path
= VirtualLinkRecord
.vlr_xpath(vlr
)
4182 return (yield from self
._vlr
_pub
_hdlr
.update(xact
, path
, vlr
))
4185 def unpublish_vlr(self
, xact
, vlr
):
4186 """ Unpublish a VLR """
4187 path
= VirtualLinkRecord
.vlr_xpath(vlr
)
4188 return (yield from self
._vlr
_pub
_hdlr
.delete(xact
, path
))
4191 class ScalingRpcHandler(mano_dts
.DtsHandler
):
4192 """ The Network service Monitor DTS handler """
4193 SCALE_IN_INPUT_XPATH
= "I,/nsr:exec-scale-in"
4194 SCALE_IN_OUTPUT_XPATH
= "O,/nsr:exec-scale-in"
4196 SCALE_OUT_INPUT_XPATH
= "I,/nsr:exec-scale-out"
4197 SCALE_OUT_OUTPUT_XPATH
= "O,/nsr:exec-scale-out"
4199 ACTION
= Enum('ACTION', 'SCALE_IN SCALE_OUT')
4201 def __init__(self
, log
, dts
, loop
, callback
=None):
4202 super().__init
__(log
, dts
, loop
)
4203 self
.callback
= callback
4204 self
.last_instance_id
= defaultdict(int)
4210 def on_scale_in_prepare(xact_info
, action
, ks_path
, msg
):
4211 assert action
== rwdts
.QueryAction
.RPC
4215 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_IN
)
4217 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleIn
.from_dict({
4218 "instance_id": msg
.instance_id
})
4220 xact_info
.respond_xpath(
4221 rwdts
.XactRspCode
.ACK
,
4222 self
.__class
__.SCALE_IN_OUTPUT_XPATH
,
4225 except Exception as e
:
4226 self
.log
.exception(e
)
4227 xact_info
.respond_xpath(
4228 rwdts
.XactRspCode
.NACK
,
4229 self
.__class
__.SCALE_IN_OUTPUT_XPATH
)
4232 def on_scale_out_prepare(xact_info
, action
, ks_path
, msg
):
4233 assert action
== rwdts
.QueryAction
.RPC
4236 scaling_group
= msg
.scaling_group_name_ref
4237 if not msg
.instance_id
:
4238 last_instance_id
= self
.last_instance_id
[scale_group
]
4239 msg
.instance_id
= last_instance_id
+ 1
4240 self
.last_instance_id
[scale_group
] += 1
4243 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_OUT
)
4245 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleOut
.from_dict({
4246 "instance_id": msg
.instance_id
})
4248 xact_info
.respond_xpath(
4249 rwdts
.XactRspCode
.ACK
,
4250 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
,
4253 except Exception as e
:
4254 self
.log
.exception(e
)
4255 xact_info
.respond_xpath(
4256 rwdts
.XactRspCode
.NACK
,
4257 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
)
4259 scale_in_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4260 on_prepare
=on_scale_in_prepare
)
4261 scale_out_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4262 on_prepare
=on_scale_out_prepare
)
4264 with self
.dts
.group_create() as group
:
4266 xpath
=self
.__class
__.SCALE_IN_INPUT_XPATH
,
4267 handler
=scale_in_hdl
,
4268 flags
=rwdts
.Flag
.PUBLISHER
)
4270 xpath
=self
.__class
__.SCALE_OUT_INPUT_XPATH
,
4271 handler
=scale_out_hdl
,
4272 flags
=rwdts
.Flag
.PUBLISHER
)
4275 class NsmTasklet(rift
.tasklets
.Tasklet
):
4277 The network service manager tasklet
4279 def __init__(self
, *args
, **kwargs
):
4280 super(NsmTasklet
, self
).__init
__(*args
, **kwargs
)
4281 self
.rwlog
.set_category("rw-mano-log")
4282 self
.rwlog
.set_subcategory("nsm")
4287 self
._ro
_plugin
_selector
= None
4288 self
._vnffgmgr
= None
4290 self
._nsr
_handler
= None
4291 self
._vnfr
_pub
_handler
= None
4292 self
._vlr
_pub
_handler
= None
4293 self
._vnfd
_pub
_handler
= None
4294 self
._scale
_cfg
_handler
= None
4296 self
._records
_publisher
_proxy
= None
4299 """ The task start callback """
4300 super(NsmTasklet
, self
).start()
4301 self
.log
.info("Starting NsmTasklet")
4303 self
.log
.debug("Registering with dts")
4304 self
._dts
= rift
.tasklets
.DTS(self
.tasklet_info
,
4305 RwNsmYang
.get_schema(),
4307 self
.on_dts_state_change
)
4309 self
.log
.debug("Created DTS Api GI Object: %s", self
._dts
)
4315 print("Caught Exception in NSM stop:", sys
.exc_info()[0])
4318 def on_instance_started(self
):
4319 """ Task instance started callback """
4320 self
.log
.debug("Got instance started callback")
4324 """ Task init callback """
4325 self
.log
.debug("Got instance started callback")
4327 self
.log
.debug("creating config account handler")
4329 self
._nsr
_pub
_handler
= publisher
.NsrOpDataDtsHandler(self
._dts
, self
.log
, self
.loop
)
4330 yield from self
._nsr
_pub
_handler
.register()
4332 self
._vnfr
_pub
_handler
= publisher
.VnfrPublisherDtsHandler(self
._dts
, self
.log
, self
.loop
)
4333 yield from self
._vnfr
_pub
_handler
.register()
4335 self
._vlr
_pub
_handler
= publisher
.VlrPublisherDtsHandler(self
._dts
, self
.log
, self
.loop
)
4336 yield from self
._vlr
_pub
_handler
.register()
4338 manifest
= self
.tasklet_info
.get_pb_manifest()
4339 use_ssl
= manifest
.bootstrap_phase
.rwsecurity
.use_ssl
4340 ssl_cert
= manifest
.bootstrap_phase
.rwsecurity
.cert
4341 ssl_key
= manifest
.bootstrap_phase
.rwsecurity
.key
4343 self
._vnfd
_pub
_handler
= publisher
.VnfdPublisher(use_ssl
, ssl_cert
, ssl_key
, self
.loop
)
4345 self
._records
_publisher
_proxy
= NsmRecordsPublisherProxy(
4349 self
._nsr
_pub
_handler
,
4350 self
._vnfr
_pub
_handler
,
4351 self
._vlr
_pub
_handler
,
4354 # Register the NSM to receive the nsm plugin
4355 # when cloud account is configured
4356 self
._ro
_plugin
_selector
= cloud
.ROAccountPluginSelector(
4360 self
._records
_publisher
_proxy
,
4362 yield from self
._ro
_plugin
_selector
.register()
4364 self
._cloud
_account
_handler
= cloud
.CloudAccountConfigSubscriber(
4369 yield from self
._cloud
_account
_handler
.register()
4371 self
._vnffgmgr
= rwvnffgmgr
.VnffgMgr(self
._dts
, self
.log
, self
.log_hdl
, self
.loop
)
4372 yield from self
._vnffgmgr
.register()
4374 self
._nsm
= NsManager(
4378 self
._nsr
_pub
_handler
,
4379 self
._vnfr
_pub
_handler
,
4380 self
._vlr
_pub
_handler
,
4381 self
._ro
_plugin
_selector
,
4383 self
._vnfd
_pub
_handler
,
4384 self
._cloud
_account
_handler
4387 yield from self
._nsm
.register()
4391 """ Task run callback """
4395 def on_dts_state_change(self
, state
):
4396 """Take action according to current dts state to transition
4397 application into the corresponding application state
4400 state - current dts state
4403 rwdts
.State
.INIT
: rwdts
.State
.REGN_COMPLETE
,
4404 rwdts
.State
.CONFIG
: rwdts
.State
.RUN
,
4408 rwdts
.State
.INIT
: self
.init
,
4409 rwdts
.State
.RUN
: self
.run
,
4412 # Transition application to next state
4413 handler
= handlers
.get(state
, None)
4414 if handler
is not None:
4415 yield from handler()
4417 # Transition dts to next state
4418 next_state
= switch
.get(state
, None)
4419 if next_state
is not None:
4420 self
.log
.debug("Changing state to %s", next_state
)
4421 self
._dts
.handle
.set_state(next_state
)