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
796 self
._prev
_state
= VnfRecordState
.INIT
797 self
._state
= VnfRecordState
.INIT
798 self
._state
_failed
_reason
= None
800 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
803 self
._vnfr
_id
= str(uuid
.uuid4())
805 self
._vnfr
_msg
= self
.create_vnfr_msg()
806 self
._log
.debug("Set VNFR {} config type to {}".
807 format(self
.name
, self
.config_type
))
808 self
.restart_mode
= restart_mode
811 if group_name
is None and group_instance_id
is not None:
812 raise ValueError("Group instance id must not be provided with an empty group name")
822 return "D,/vnfr:vnfr-catalog/vnfr:vnfr[vnfr:id = '{}']".format(self
.id)
827 return self
._vnfr
_msg
830 def const_vnfr_msg(self
):
832 return RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_ConstituentVnfrRef(vnfr_id
=self
.id,cloud_account
=self
.cloud_account_name
,om_datacenter
=self
._om
_datacenter
_name
)
840 def cloud_account_name(self
):
841 """ Cloud account that this VNF should be created in """
842 return self
._cloud
_account
_name
845 def om_datacenter_name(self
):
846 """ Datacenter that this VNF should be created in """
847 return self
._om
_datacenter
_name
852 """ Is this VNF actve """
853 return True if self
._state
== VnfRecordState
.ACTIVE
else False
857 """ state of this VNF """
861 def state_failed_reason(self
):
862 """ Error message in case this VNF is in failed state """
863 return self
._state
_failed
_reason
866 def member_vnf_index(self
):
867 """ Member VNF index """
868 return self
._const
_vnfd
_msg
.member_vnf_index
873 return self
._nsr
_name
877 """ Name of this VNFR """
878 if self
._name
is not None:
881 name_tags
= [self
._nsr
_name
]
883 if self
._group
_name
is not None:
884 name_tags
.append(self
._group
_name
)
886 if self
._group
_instance
_id
is not None:
887 name_tags
.append(str(self
._group
_instance
_id
))
889 name_tags
.extend([self
.vnfd
.name
, str(self
.member_vnf_index
)])
891 self
._name
= "__".join(name_tags
)
896 def vnfr_xpath(vnfr
):
897 """ Get the VNFR path from VNFR """
898 return (VirtualNetworkFunctionRecord
.XPATH
+ "[vnfr:id = '{}']").format(vnfr
.id)
901 def config_type(self
):
902 cfg_types
= ['netconf', 'juju', 'script']
903 for method
in cfg_types
:
904 if self
._vnfd
.vnf_configuration
.has_field(method
):
909 def config_status(self
):
910 """Return the config status as YANG ENUM string"""
911 self
._log
.debug("Map VNFR {} config status {} ({})".
912 format(self
.name
, self
._config
_status
, self
.config_type
))
913 if self
.config_type
== 'none':
914 return 'config_not_needed'
915 elif self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
917 elif self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
922 def set_state(self
, state
):
923 """ set the state of this object """
924 self
._prev
_state
= self
._state
927 def reset_id(self
, vnfr_id
):
928 self
._vnfr
_id
= vnfr_id
929 self
._vnfr
_msg
= self
.create_vnfr_msg()
932 self
.config_store
.merge_vnfd_config(
935 self
.member_vnf_index
,
938 def create_vnfr_msg(self
):
939 """ VNFR message for this VNFR """
947 vnfd_copy_dict
= {k
: v
for k
, v
in self
._vnfd
.as_dict().items() if k
in vnfd_fields
}
950 "nsr_id_ref": self
._nsr
_id
,
951 "vnfd_ref": self
.vnfd
.id,
953 "cloud_account": self
._cloud
_account
_name
,
954 "om_datacenter": self
._om
_datacenter
_name
,
955 "config_status": self
.config_status
957 vnfr_dict
.update(vnfd_copy_dict
)
959 vnfr
= RwVnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr
.from_dict(vnfr_dict
)
960 vnfr
.member_vnf_index_ref
= self
.member_vnf_index
961 vnfr
.vnf_configuration
.from_dict(self
._vnfd
.vnf_configuration
.as_dict())
963 if self
._vnfd
.mgmt_interface
.has_field("port"):
964 vnfr
.mgmt_interface
.port
= self
._vnfd
.mgmt_interface
.port
966 for group_info
in self
._placement
_groups
:
967 group
= vnfr
.placement_groups_info
.add()
968 group
.from_dict(group_info
.as_dict())
970 # UI expects the monitoring param field to exist
971 vnfr
.monitoring_param
= []
973 self
._log
.debug("Get vnfr_msg for VNFR {} : {}".format(self
.name
, vnfr
))
977 def update_vnfm(self
):
978 self
._log
.debug("Send an update to VNFM for VNFR {} with {}".
979 format(self
.name
, self
.vnfr_msg
))
980 yield from self
._dts
.query_update(
982 rwdts
.XactFlag
.TRACE
,
986 def get_config_status(self
):
987 """Return the config status as YANG ENUM"""
988 return self
._config
_status
991 def set_config_status(self
, status
):
993 def status_to_string(status
):
995 NsrYang
.ConfigStates
.INIT
: 'init',
996 NsrYang
.ConfigStates
.CONFIGURING
: 'configuring',
997 NsrYang
.ConfigStates
.CONFIG_NOT_NEEDED
: 'config_not_needed',
998 NsrYang
.ConfigStates
.CONFIGURED
: 'configured',
999 NsrYang
.ConfigStates
.FAILED
: 'failed',
1002 return status_dc
[status
]
1004 self
._log
.debug("Update VNFR {} from {} ({}) to {}".
1005 format(self
.name
, self
._config
_status
,
1006 self
.config_type
, status
))
1007 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1008 self
._log
.error("Updating already configured VNFR {}".
1012 if self
._config
_status
!= status
:
1014 self
._config
_status
= status
1015 # I don't think this is used. Original implementor can check.
1016 # Caused Exception, so corrected it by status_to_string
1017 # But not sure whats the use of this variable?
1018 self
.vnfr_msg
.config_status
= status_to_string(status
)
1019 except Exception as e
:
1020 self
._log
.error("Exception=%s", str(e
))
1023 self
._log
.debug("Updated VNFR {} status to {}".format(self
.name
, status
))
1025 if self
._config
_status
!= NsrYang
.ConfigStates
.INIT
:
1027 # Publish only after VNFM has the VNFR created
1028 yield from self
.update_vnfm()
1029 except Exception as e
:
1030 self
._log
.error("Exception updating VNFM with new status {} of VNFR {}: {}".
1031 format(status
, self
.name
, e
))
1032 self
._log
.exception(e
)
1034 def is_configured(self
):
1035 if self
.config_type
== 'none':
1038 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1044 def instantiate(self
, nsr
):
1045 """ Instantiate this VNFR"""
1047 self
._log
.debug("Instaniating VNFR key %s, vnfd %s",
1048 self
.xpath
, self
._vnfd
)
1050 self
._log
.debug("Create VNF with xpath %s and vnfr %s",
1051 self
.xpath
, self
.vnfr_msg
)
1053 self
.set_state(VnfRecordState
.INSTANTIATION_PENDING
)
1055 def find_vlr_for_cp(conn
):
1056 """ Find VLR for the given connection point """
1057 for vlr
in nsr
.vlrs
:
1058 for vnfd_cp
in vlr
.vld_msg
.vnfd_connection_point_ref
:
1059 if (vnfd_cp
.vnfd_id_ref
== self
._vnfd
.id and
1060 vnfd_cp
.vnfd_connection_point_ref
== conn
.name
and
1061 vnfd_cp
.member_vnf_index_ref
== self
.member_vnf_index
and
1062 vlr
.cloud_account_name
== self
.cloud_account_name
):
1063 self
._log
.debug("Found VLR for cp_name:%s and vnf-index:%d",
1064 conn
.name
, self
.member_vnf_index
)
1068 # For every connection point in the VNFD fill in the identifier
1069 for conn_p
in self
._vnfd
.connection_point
:
1070 cpr
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_ConnectionPoint()
1071 cpr
.name
= conn_p
.name
1072 cpr
.type_yang
= conn_p
.type_yang
1073 vlr_ref
= find_vlr_for_cp(conn_p
)
1075 msg
= "Failed to find VLR for cp = %s" % conn_p
.name
1076 self
._log
.debug("%s", msg
)
1077 # raise VirtualNetworkFunctionRecordError(msg)
1080 cpr
.vlr_ref
= vlr_ref
.id
1081 self
.vnfr_msg
.connection_point
.append(cpr
)
1082 self
._log
.debug("Connection point [%s] added, vnf id=%s vnfd id=%s",
1083 cpr
, self
.vnfr_msg
.id, self
.vnfr_msg
.vnfd_ref
)
1085 if not self
.restart_mode
:
1086 yield from self
._dts
.query_create(self
.xpath
,
1090 yield from self
._dts
.query_update(self
.xpath
,
1094 self
._log
.info("Created VNF with xpath %s and vnfr %s",
1095 self
.xpath
, self
.vnfr_msg
)
1097 self
._log
.info("Instantiated VNFR with xpath %s and vnfd %s, vnfr %s",
1098 self
.xpath
, self
._vnfd
, self
.vnfr_msg
)
1101 def update_state(self
, vnfr_msg
):
1102 """ Update this VNFR"""
1103 if vnfr_msg
.operational_status
== "running":
1104 if self
.vnfr_msg
.operational_status
!= "running":
1105 yield from self
.is_active()
1106 elif vnfr_msg
.operational_status
== "failed":
1107 yield from self
.instantiation_failed(failed_reason
=vnfr_msg
.operational_status_details
)
1110 def is_active(self
):
1111 """ This VNFR is active """
1112 self
._log
.debug("VNFR %s is active", self
._vnfr
_id
)
1113 self
.set_state(VnfRecordState
.ACTIVE
)
1116 def instantiation_failed(self
, failed_reason
=None):
1117 """ This VNFR instantiation failed"""
1118 self
._log
.error("VNFR %s instantiation failed", self
._vnfr
_id
)
1119 self
.set_state(VnfRecordState
.FAILED
)
1120 self
._state
_failed
_reason
= failed_reason
1122 def vnfr_in_vnfm(self
):
1123 """ Is there a VNFR record in VNFM """
1124 if (self
._state
== VnfRecordState
.ACTIVE
or
1125 self
._state
== VnfRecordState
.INSTANTIATION_PENDING
or
1126 self
._state
== VnfRecordState
.FAILED
):
1132 def terminate(self
):
1133 """ Terminate this VNF """
1134 if not self
.vnfr_in_vnfm():
1135 self
._log
.debug("Ignoring terminate request for id %s in state %s",
1136 self
.id, self
._state
)
1139 self
._log
.debug("Terminating VNF id:%s", self
.id)
1140 self
.set_state(VnfRecordState
.TERMINATE_PENDING
)
1141 with self
._dts
.transaction(flags
=0) as xact
:
1142 block
= xact
.block_create()
1143 block
.add_query_delete(self
.xpath
)
1144 yield from block
.execute(flags
=0)
1145 self
.set_state(VnfRecordState
.TERMINATED
)
1146 self
._log
.debug("Terminated VNF id:%s", self
.id)
1149 class NetworkServiceStatus(object):
1150 """ A class representing the Network service's status """
1151 MAX_EVENTS_RECORDED
= 10
1152 """ Network service Status class"""
1153 def __init__(self
, dts
, log
, loop
):
1158 self
._state
= NetworkServiceRecordState
.INIT
1159 self
._events
= deque([])
1162 def create_notification(self
, evt
, evt_desc
, evt_details
):
1163 xp
= "N,/rw-nsr:nsm-notification"
1164 notif
= RwNsrYang
.YangNotif_RwNsr_NsmNotification()
1166 notif
.description
= evt_desc
1167 notif
.details
= evt_details
if evt_details
is not None else None
1169 yield from self
._dts
.query_create(xp
, rwdts
.XactFlag
.ADVISE
, notif
)
1170 self
._log
.info("Notification called by creating dts query: %s", notif
)
1172 def record_event(self
, evt
, evt_desc
, evt_details
):
1173 """ Record an event """
1174 self
._log
.debug("Recording event - evt %s, evt_descr %s len = %s",
1175 evt
, evt_desc
, len(self
._events
))
1176 if len(self
._events
) >= NetworkServiceStatus
.MAX_EVENTS_RECORDED
:
1177 self
._events
.popleft()
1178 self
._events
.append((int(time
.time()), evt
, evt_desc
,
1179 evt_details
if evt_details
is not None else None))
1181 self
._loop
.create_task(self
.create_notification(evt
,evt_desc
,evt_details
))
1183 def set_state(self
, state
):
1184 """ set the state of this status object """
1188 """ Return the state as a yang enum string """
1189 state_to_str_map
= {"INIT": "init",
1190 "VL_INIT_PHASE": "vl_init_phase",
1191 "VNF_INIT_PHASE": "vnf_init_phase",
1192 "VNFFG_INIT_PHASE": "vnffg_init_phase",
1193 "SCALING_GROUP_INIT_PHASE": "scaling_group_init_phase",
1194 "RUNNING": "running",
1195 "SCALING_OUT": "scaling_out",
1196 "SCALING_IN": "scaling_in",
1197 "TERMINATE_RCVD": "terminate_rcvd",
1198 "TERMINATE": "terminate",
1199 "VL_TERMINATE_PHASE": "vl_terminate_phase",
1200 "VNF_TERMINATE_PHASE": "vnf_terminate_phase",
1201 "VNFFG_TERMINATE_PHASE": "vnffg_terminate_phase",
1202 "TERMINATED": "terminated",
1204 "VL_INSTANTIATE": "vl_instantiate",
1205 "VL_TERMINATE": "vl_terminate",
1207 return state_to_str_map
[self
._state
.name
]
1211 """ State of this status object """
1216 """ Network Service Record as a message"""
1219 for entry
in self
._events
:
1220 event
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_OperationalEvents()
1223 event
.timestamp
, event
.event
, event
.description
, event
.details
= entry
1224 event_list
.append(event
)
1228 class NetworkServiceRecord(object):
1229 """ Network service record """
1230 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
1232 def __init__(self
, dts
, log
, loop
, nsm
, nsm_plugin
, nsr_cfg_msg
, sdn_account_name
, key_pairs
, restart_mode
=False,
1238 self
._nsr
_cfg
_msg
= nsr_cfg_msg
1239 self
._nsm
_plugin
= nsm_plugin
1240 self
._sdn
_account
_name
= sdn_account_name
1241 self
._vlr
_handler
= vlr_handler
1244 self
._nsr
_msg
= None
1245 self
._nsr
_regh
= None
1246 self
._key
_pairs
= key_pairs
1251 self
._param
_pools
= {}
1252 self
._scaling
_groups
= {}
1253 self
._create
_time
= int(time
.time())
1254 self
._op
_status
= NetworkServiceStatus(dts
, log
, loop
)
1255 self
._config
_status
= NsrYang
.ConfigStates
.CONFIGURING
1256 self
._config
_status
_details
= None
1258 self
.restart_mode
= restart_mode
1259 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
1260 self
._debug
_running
= False
1261 self
._is
_active
= False
1262 self
._vl
_phase
_completed
= False
1263 self
._vnf
_phase
_completed
= False
1266 # Initalise the state to init
1267 # The NSR moves through the following transitions
1268 # 1. INIT -> VLS_READY once all the VLs in the NSD are created
1269 # 2. VLS_READY - VNFS_READY when all the VNFs in the NSD are created
1270 # 3. VNFS_READY - READY when the NSR is published
1272 self
.set_state(NetworkServiceRecordState
.INIT
)
1274 self
.substitute_input_parameters
= InputParameterSubstitution(self
._log
)
1277 def nsm_plugin(self
):
1279 return self
._nsm
_plugin
1281 def set_state(self
, state
):
1282 """ Set state for this NSR"""
1283 self
._log
.debug("Setting state to %s", state
)
1284 # We are in init phase and is moving to the next state
1285 # The new state could be a FAILED state or VNF_INIIT_PHASE
1286 if self
.state
== NetworkServiceRecordState
.VL_INIT_PHASE
:
1287 self
._vl
_phase
_completed
= True
1289 if self
.state
== NetworkServiceRecordState
.VNF_INIT_PHASE
:
1290 self
._vnf
_phase
_completed
= True
1292 self
._op
_status
.set_state(state
)
1296 """ Get id for this NSR"""
1297 return self
._nsr
_cfg
_msg
.id
1301 """ Name of this network service record """
1302 return self
._nsr
_cfg
_msg
.name
1305 def cloud_account_name(self
):
1306 return self
._nsr
_cfg
_msg
.cloud_account
1309 def om_datacenter_name(self
):
1310 if self
._nsr
_cfg
_msg
.has_field('om_datacenter'):
1311 return self
._nsr
_cfg
_msg
.om_datacenter
1316 """State of this NetworkServiceRecord"""
1317 return self
._op
_status
.state
1321 """ Is this NSR active ?"""
1322 return True if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
else False
1326 """ VLRs associated with this NSR"""
1331 """ VNFRs associated with this NSR"""
1336 """ VNFFGRs associated with this NSR"""
1337 return self
._vnffgrs
1340 def scaling_groups(self
):
1341 """ Scaling groups associated with this NSR """
1342 return self
._scaling
_groups
1345 def param_pools(self
):
1346 """ Parameter value pools associated with this NSR"""
1347 return self
._param
_pools
1350 def nsr_cfg_msg(self
):
1351 return self
._nsr
_cfg
_msg
1354 def nsr_cfg_msg(self
, msg
):
1355 self
._nsr
_cfg
_msg
= msg
1359 """ NSD Protobuf for this NSR """
1360 if self
._nsd
is not None:
1362 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
1367 """ NSD ID for this NSR """
1368 return self
.nsd_msg
.id
1372 ''' Get a new job id for config primitive'''
1377 def config_status(self
):
1378 """ Config status for NSR """
1379 return self
._config
_status
1381 def resolve_placement_group_cloud_construct(self
, input_group
):
1383 Returns the cloud specific construct for placement group
1385 copy_dict
= ['name', 'requirement', 'strategy']
1387 for group_info
in self
._nsr
_cfg
_msg
.nsd_placement_group_maps
:
1388 if group_info
.placement_group_ref
== input_group
.name
:
1389 group
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_PlacementGroupsInfo()
1390 group_dict
= {k
:v
for k
,v
in
1391 group_info
.as_dict().items() if k
!= 'placement_group_ref'}
1392 for param
in copy_dict
:
1393 group_dict
.update({param
: getattr(input_group
, param
)})
1394 group
.from_dict(group_dict
)
1400 return "NSR(name={}, nsd_id={}, cloud_account={})".format(
1401 self
.name
, self
.nsd_id
, self
.cloud_account_name
1404 def _get_vnfd(self
, vnfd_id
, config_xact
):
1405 """ Fetch vnfd msg for the passed vnfd id """
1406 return self
._nsm
.get_vnfd(vnfd_id
, config_xact
)
1408 def _get_vnfd_cloud_account(self
, vnfd_member_index
):
1409 """ Fetch Cloud Account for the passed vnfd id """
1410 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1411 vim_accounts
= [(vnf
.cloud_account
,vnf
.om_datacenter
) for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map \
1412 if vnfd_member_index
== vnf
.member_vnf_index_ref
]
1413 if vim_accounts
and vim_accounts
[0]:
1414 return vim_accounts
[0]
1415 return (self
.cloud_account_name
,self
.om_datacenter_name
)
1417 def _get_constituent_vnfd_msg(self
, vnf_index
):
1418 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1419 if const_vnfd
.member_vnf_index
== vnf_index
:
1422 raise ValueError("Constituent VNF index %s not found" % vnf_index
)
1424 def record_event(self
, evt
, evt_desc
, evt_details
=None, state
=None):
1425 """ Record an event """
1426 self
._op
_status
.record_event(evt
, evt_desc
, evt_details
)
1427 if state
is not None:
1428 self
.set_state(state
)
1430 def scaling_trigger_str(self
, trigger
):
1431 SCALING_TRIGGER_STRS
= {
1432 NsdYang
.ScalingTrigger
.PRE_SCALE_IN
: 'pre-scale-in',
1433 NsdYang
.ScalingTrigger
.POST_SCALE_IN
: 'post-scale-in',
1434 NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
: 'pre-scale-out',
1435 NsdYang
.ScalingTrigger
.POST_SCALE_OUT
: 'post-scale-out',
1438 return SCALING_TRIGGER_STRS
[trigger
]
1439 except Exception as e
:
1440 self
._log
.error("Scaling trigger mapping error for {} : {}".
1442 self
._log
.exception(e
)
1443 return "Unknown trigger"
1446 def instantiate_vls(self
):
1448 This function instantiates VLs for every VL in this Network Service
1450 self
._log
.debug("Instantiating %d VLs in NSD id %s", len(self
._vlrs
),
1452 for vlr
in self
._vlrs
:
1453 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1454 vlr
.state
= VlRecordState
.ACTIVE
1455 self
._loop
.create_task(self
.vlr_uptime_update(vlr
))
1458 def vlr_uptime_update(self
, vlr
):
1459 vlr_
= RwVlrYang
.YangData_Vlr_VlrCatalog_Vlr
.from_dict({'id': vlr
.id})
1461 if vlr
.state
not in [VlRecordState
.INIT
, VlRecordState
.INSTANTIATION_PENDING
, VlRecordState
.ACTIVE
]:
1463 vlr_
.uptime
= int(time
.time()) - vlr
._create
_time
1464 yield from self
._vlr
_handler
.update(None, VirtualLinkRecord
.vlr_xpath(vlr
), vlr_
)
1465 yield from asyncio
.sleep(2, loop
=self
._loop
)
1469 def create(self
, config_xact
):
1470 """ Create this network service"""
1471 # Create virtual links for all the external vnf
1472 # connection points in this NS
1473 yield from self
.create_vls()
1475 # Create VNFs in this network service
1476 yield from self
.create_vnfs(config_xact
)
1478 # Create VNFFG for network service
1479 self
.create_vnffgs()
1481 # Create Scaling Groups for each scaling group in NSD
1482 self
.create_scaling_groups()
1484 # Create Parameter Pools
1485 self
.create_param_pools()
1488 def apply_scale_group_config_script(self
, script
, group
, scale_instance
, trigger
, vnfrs
=None):
1489 """ Apply config based on script for scale group """
1492 def add_vnfrs_data(vnfrs_list
):
1493 """ Add as a dict each of the VNFRs data """
1495 for vnfr
in vnfrs_list
:
1496 self
._log
.debug("Add VNFR {} data".format(vnfr
))
1498 vnfr_data
['name'] = vnfr
.name
1499 if trigger
in [NsdYang
.ScalingTrigger
.PRE_SCALE_IN
, NsdYang
.ScalingTrigger
.POST_SCALE_OUT
]:
1500 # Get VNF management and other IPs, etc
1501 opdata
= yield from self
.fetch_vnfr(vnfr
.xpath
)
1502 self
._log
.debug("VNFR {} op data: {}".format(vnfr
.name
, opdata
))
1504 vnfr_data
['rw_mgmt_ip'] = opdata
.mgmt_interface
.ip_address
1505 vnfr_data
['rw_mgmt_port'] = opdata
.mgmt_interface
.port
1506 except Exception as e
:
1507 self
._log
.error("Unable to get management IP for vnfr {}:{}".
1508 format(vnfr
.name
, e
))
1511 vnfr_data
['connection_points'] = []
1512 for cp
in opdata
.connection_point
:
1514 con_pt
['name'] = cp
.name
1515 con_pt
['ip_address'] = cp
.ip_address
1516 vnfr_data
['connection_points'].append(con_pt
)
1517 except Exception as e
:
1518 self
._log
.error("Exception getting connections points for VNFR {}: {}".
1519 format(vnfr
.name
, e
))
1521 vnfrs_data
.append(vnfr_data
)
1522 self
._log
.debug("VNFRs data: {}".format(vnfrs_data
))
1526 def add_nsr_data(nsr
):
1528 nsr_data
['name'] = nsr
.name
1531 if script
is None or len(script
) == 0:
1532 self
._log
.error("Script not provided for scale group config: {}".format(group
.name
))
1535 if script
[0] == '/':
1538 path
= os
.path
.join(os
.environ
['RIFT_INSTALL'], "usr/bin", script
)
1539 if not os
.path
.exists(path
):
1540 self
._log
.error("Config faled for scale group {}: Script does not exist at {}".
1541 format(group
.name
, path
))
1544 # Build a YAML file with all parameters for the script to execute
1545 # The data consists of 5 sections
1547 # 2. Scale group config
1548 # 3. VNFRs in the scale group
1549 # 4. VNFRs outside scale group
1552 data
['trigger'] = group
.trigger_map(trigger
)
1553 data
['config'] = group
.group_msg
.as_dict()
1556 data
["vnfrs_in_group"] = yield from add_vnfrs_data(vnfrs
)
1558 data
["vnfrs_in_group"] = yield from add_vnfrs_data(scale_instance
.vnfrs
)
1560 data
["vnfrs_others"] = yield from add_vnfrs_data(self
.vnfrs
.values())
1561 data
["nsr"] = add_nsr_data(self
)
1564 with tempfile
.NamedTemporaryFile(delete
=False) as tmp_file
:
1565 tmp_file
.write(yaml
.dump(data
, default_flow_style
=True)
1568 self
._log
.debug("Creating a temp file: {} with input data: {}".
1569 format(tmp_file
.name
, data
))
1571 cmd
= "{} {}".format(path
, tmp_file
.name
)
1572 self
._log
.debug("Running the CMD: {}".format(cmd
))
1573 proc
= yield from asyncio
.create_subprocess_shell(cmd
, loop
=self
._loop
)
1574 rc
= yield from proc
.wait()
1576 self
._log
.error("The script {} for scale group {} config returned: {}".
1577 format(script
, group
.name
, rc
))
1585 def apply_scaling_group_config(self
, trigger
, group
, scale_instance
, vnfrs
=None):
1586 """ Apply the config for the scaling group based on trigger """
1587 if group
is None or scale_instance
is None:
1591 def update_config_status(success
=True, err_msg
=None):
1592 self
._log
.debug("Update %s config status to %r : %s",
1593 scale_instance
, success
, err_msg
)
1594 if (scale_instance
.config_status
== "failed"):
1595 # Do not update the config status if it is already in failed state
1598 if scale_instance
.config_status
== "configured":
1599 # Update only to failed state an already configured scale instance
1601 scale_instance
.config_status
= "failed"
1602 scale_instance
.config_err_msg
= err_msg
1603 yield from self
.update_state()
1605 # We are in configuring state
1606 # Only after post scale out mark instance as configured
1607 if trigger
== NsdYang
.ScalingTrigger
.POST_SCALE_OUT
:
1609 scale_instance
.config_status
= "configured"
1611 scale_instance
.config_status
= "failed"
1612 scale_instance
.config_err_msg
= err_msg
1613 yield from self
.update_state()
1615 config
= group
.trigger_config(trigger
)
1619 self
._log
.debug("Scaling group {} config: {}".format(group
.name
, config
))
1620 if config
.has_field("ns_config_primitive_name_ref"):
1621 config_name
= config
.ns_config_primitive_name_ref
1622 nsd_msg
= self
.nsd_msg
1623 config_primitive
= None
1624 for ns_cfg_prim
in nsd_msg
.service_primitive
:
1625 if ns_cfg_prim
.name
== config_name
:
1626 config_primitive
= ns_cfg_prim
1629 if config_primitive
is None:
1630 raise ValueError("Could not find ns_cfg_prim %s in nsr %s" % (config_name
, self
.name
))
1632 self
._log
.debug("Scaling group {} config primitive: {}".format(group
.name
, config_primitive
))
1633 if config_primitive
.has_field("user_defined_script"):
1634 rc
= yield from self
.apply_scale_group_config_script(config_primitive
.user_defined_script
,
1635 group
, scale_instance
, trigger
, vnfrs
)
1638 err_msg
= "Failed config for trigger {} using config script '{}'". \
1639 format(self
.scaling_trigger_str(trigger
),
1640 config_primitive
.user_defined_script
)
1641 yield from update_config_status(success
=rc
, err_msg
=err_msg
)
1644 err_msg
= "Failed config for trigger {} as config script is not specified". \
1645 format(self
.scaling_trigger_str(trigger
))
1646 yield from update_config_status(success
=False, err_msg
=err_msg
)
1647 raise NotImplementedError("Only script based config support for scale group for now: {}".
1650 err_msg
= "Failed config for trigger {} as config primitive is not specified".\
1651 format(self
.scaling_trigger_str(trigger
))
1652 yield from update_config_status(success
=False, err_msg
=err_msg
)
1653 self
._log
.error("Config primitive not specified for config action in scale group %s" %
1657 def create_scaling_groups(self
):
1658 """ This function creates a NSScalingGroup for every scaling
1659 group defined in he NSD"""
1661 for scaling_group_msg
in self
.nsd_msg
.scaling_group_descriptor
:
1662 self
._log
.debug("Found scaling_group %s in nsr id %s",
1663 scaling_group_msg
.name
, self
.id)
1665 group_record
= scale_group
.ScalingGroup(
1670 self
._scaling
_groups
[group_record
.name
] = group_record
1673 def create_scale_group_instance(self
, group_name
, index
, config_xact
, is_default
=False):
1674 group
= self
._scaling
_groups
[group_name
]
1675 scale_instance
= group
.create_instance(index
, is_default
)
1679 self
._log
.debug("Creating %u VNFs associated with NS id %s scaling group %s",
1680 len(self
.nsd_msg
.constituent_vnfd
), self
.id, self
)
1683 for vnf_index
, count
in group
.vnf_index_count_map
.items():
1684 const_vnfd_msg
= self
._get
_constituent
_vnfd
_msg
(vnf_index
)
1685 vnfd_msg
= self
._get
_vnfd
(const_vnfd_msg
.vnfd_id_ref
, config_xact
)
1687 cloud_account_name
, om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd_msg
.member_vnf_index
)
1688 if cloud_account_name
is None:
1689 cloud_account_name
= self
.cloud_account_name
1690 for _
in range(count
):
1691 vnfr
= yield from self
.create_vnf_record(vnfd_msg
, const_vnfd_msg
, cloud_account_name
, om_datacenter_name
, group_name
, index
)
1692 scale_instance
.add_vnfr(vnfr
)
1698 def instantiate_instance():
1699 self
._log
.debug("Creating %s VNFRS", scale_instance
)
1700 vnfrs
= yield from create_vnfs()
1701 yield from self
.publish()
1703 self
._log
.debug("Instantiating %s VNFRS for %s", len(vnfrs
), scale_instance
)
1704 scale_instance
.operational_status
= "vnf_init_phase"
1705 yield from self
.update_state()
1708 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
,
1709 group
, scale_instance
, vnfrs
)
1711 self
._log
.error("Pre scale out config for scale group {} ({}) failed".
1712 format(group
.name
, index
))
1713 scale_instance
.operational_status
= "failed"
1715 yield from self
.instantiate_vnfs(vnfrs
)
1717 except Exception as e
:
1718 self
._log
.exception("Failed to begin instantiatiation of vnfs for scale group {}: {}".
1719 format(group
.name
, e
))
1720 self
._log
.exception(e
)
1721 scale_instance
.operational_status
= "failed"
1723 yield from self
.update_state()
1725 yield from instantiate_instance()
1728 def delete_scale_group_instance(self
, group_name
, index
):
1729 group
= self
._scaling
_groups
[group_name
]
1730 scale_instance
= group
.get_instance(index
)
1731 if scale_instance
.is_default
:
1732 raise ScalingOperationError("Cannot terminate a default scaling group instance")
1734 scale_instance
.operational_status
= "terminate"
1735 yield from self
.update_state()
1738 def terminate_instance():
1739 self
._log
.debug("Terminating %s VNFRS" % scale_instance
)
1740 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_IN
,
1741 group
, scale_instance
)
1743 self
._log
.error("Pre scale in config for scale group {} ({}) failed".
1744 format(group
.name
, index
))
1746 # Going ahead with terminate, even if there is an error in pre-scale-in config
1747 # as this could be result of scale out failure and we need to cleanup this group
1748 yield from self
.terminate_vnfrs(scale_instance
.vnfrs
)
1749 group
.delete_instance(index
)
1751 scale_instance
.operational_status
= "vnf_terminate_phase"
1752 yield from self
.update_state()
1754 yield from terminate_instance()
1757 def _update_scale_group_instances_status(self
):
1759 def post_scale_out_task(group
, instance
):
1760 # Apply post scale out config once all VNFRs are active
1761 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_OUT
,
1763 instance
.operational_status
= "running"
1765 self
._log
.debug("Scale out for group {} and instance {} succeeded".
1766 format(group
.name
, instance
.instance_id
))
1768 self
._log
.error("Post scale out config for scale group {} ({}) failed".
1769 format(group
.name
, instance
.instance_id
))
1771 yield from self
.update_state()
1773 group_instances
= {group
: group
.instances
for group
in self
._scaling
_groups
.values()}
1774 for group
, instances
in group_instances
.items():
1775 self
._log
.debug("Updating %s instance status", group
)
1776 for instance
in instances
:
1777 instance_vnf_state_list
= [vnfr
.state
for vnfr
in instance
.vnfrs
]
1778 self
._log
.debug("Got vnfr instance states: %s", instance_vnf_state_list
)
1779 if instance
.operational_status
== "vnf_init_phase":
1780 if all([state
== VnfRecordState
.ACTIVE
for state
in instance_vnf_state_list
]):
1781 instance
.operational_status
= "running"
1783 # Create a task for post scale out to allow us to sleep before attempting
1784 # to configure newly created VM's
1785 self
._loop
.create_task(post_scale_out_task(group
, instance
))
1787 elif any([state
== VnfRecordState
.FAILED
for state
in instance_vnf_state_list
]):
1788 self
._log
.debug("Scale out for group {} and instance {} failed".
1789 format(group
.name
, instance
.instance_id
))
1790 instance
.operational_status
= "failed"
1792 elif instance
.operational_status
== "vnf_terminate_phase":
1793 if all([state
== VnfRecordState
.TERMINATED
for state
in instance_vnf_state_list
]):
1794 instance
.operational_status
= "terminated"
1795 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_IN
,
1798 self
._log
.debug("Scale in for group {} and instance {} succeeded".
1799 format(group
.name
, instance
.instance_id
))
1801 self
._log
.error("Post scale in config for scale group {} ({}) failed".
1802 format(group
.name
, instance
.instance_id
))
1804 def create_vnffgs(self
):
1805 """ This function creates VNFFGs for every VNFFG in the NSD
1806 associated with this NSR"""
1808 for vnffgd
in self
.nsd_msg
.vnffgd
:
1809 self
._log
.debug("Found vnffgd %s in nsr id %s", vnffgd
, self
.id)
1810 vnffgr
= VnffgRecord(self
._dts
,
1813 self
._nsm
._vnffgmgr
,
1817 self
._sdn
_account
_name
1819 self
._vnffgrs
[vnffgr
.id] = vnffgr
1821 def resolve_vld_ip_profile(self
, nsd_msg
, vld
):
1822 self
._log
.debug("Receieved ip profile ref is %s",vld
.ip_profile_ref
)
1823 if not vld
.has_field('ip_profile_ref'):
1825 profile
= [profile
for profile
in nsd_msg
.ip_profiles
if profile
.name
== vld
.ip_profile_ref
]
1826 return profile
[0] if profile
else None
1829 def _create_vls(self
, vld
, cloud_account
,om_datacenter
):
1830 """Create a VLR in the cloud account specified using the given VLD
1834 cloud_account : Cloud account name
1839 vlr
= yield from VirtualLinkRecord
.create_record(
1847 self
.resolve_vld_ip_profile(self
.nsd_msg
, vld
),
1849 restart_mode
=self
.restart_mode
)
1853 def _extract_cloud_accounts_for_vl(self
, vld
):
1855 Extracts the list of cloud accounts from the NS Config obj
1858 1. Cloud accounts based connection point (vnf_cloud_account_map)
1860 vld : VLD yang object
1865 cloud_account_list
= []
1867 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1868 # Handle case where cloud_account is None
1870 for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1871 if vnf
.cloud_account
is not None or vnf
.om_datacenter
is not None:
1872 vnf_cloud_map
[vnf
.member_vnf_index_ref
] = (vnf
.cloud_account
,vnf
.om_datacenter
)
1874 for vnfc
in vld
.vnfd_connection_point_ref
:
1875 cloud_account
= vnf_cloud_map
.get(
1876 vnfc
.member_vnf_index_ref
,
1877 (self
.cloud_account_name
,self
.om_datacenter_name
))
1879 cloud_account_list
.append(cloud_account
)
1881 if self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1882 for vld_map
in self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1883 if vld_map
.vld_id_ref
== vld
.id:
1884 for cloud_account
in vld_map
.cloud_accounts
:
1885 cloud_account_list
.extend((cloud_account
,None))
1886 for om_datacenter
in vld_map
.om_datacenters
:
1887 cloud_account_list
.extend((None,om_datacenter
))
1889 # If no config has been provided then fall-back to the default
1891 if not cloud_account_list
:
1892 cloud_account_list
= [(self
.cloud_account_name
,self
.om_datacenter_name
)]
1894 self
._log
.debug("VL {} cloud accounts: {}".
1895 format(vld
.name
, cloud_account_list
))
1896 return set(cloud_account_list
)
1899 def create_vls(self
):
1900 """ This function creates VLs for every VLD in the NSD
1901 associated with this NSR"""
1902 for vld
in self
.nsd_msg
.vld
:
1903 self
._log
.debug("Found vld %s in nsr id %s", vld
, self
.id)
1904 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1905 for cloud_account
,om_datacenter
in cloud_account_list
:
1906 vlr
= yield from self
._create
_vls
(vld
, cloud_account
,om_datacenter
)
1907 self
._vlrs
.append(vlr
)
1911 def create_vl_instance(self
, vld
):
1912 self
._log
.debug("Create VL for {}: {}".format(self
.id, vld
.as_dict()))
1913 # Check if the VL is already present
1915 for vl
in self
._vlrs
:
1916 if vl
.vld_msg
.id == vld
.id:
1917 self
._log
.debug("The VLD %s already in NSR %s as VLR %s with status %s",
1918 vld
.id, self
.id, vl
.id, vl
.state
)
1920 if vlr
.state
!= VlRecordState
.TERMINATED
:
1921 err_msg
= "VLR for VL %s in NSR %s already instantiated", \
1923 self
._log
.error(err_msg
)
1924 raise NsrVlUpdateError(err_msg
)
1928 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1929 for account
,om_datacenter
in cloud_account_list
:
1930 vlr
= yield from self
._create
_vls
(vld
, account
,om_datacenter
)
1931 self
._vlrs
.append(vlr
)
1933 vlr
.state
= VlRecordState
.INSTANTIATION_PENDING
1934 yield from self
.update_state()
1937 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1938 vlr
.state
= VlRecordState
.ACTIVE
1940 except Exception as e
:
1941 err_msg
= "Error instantiating VL for NSR {} and VLD {}: {}". \
1942 format(self
.id, vld
.id, e
)
1943 self
._log
.error(err_msg
)
1944 self
._log
.exception(e
)
1945 vlr
.state
= VlRecordState
.FAILED
1947 yield from self
.update_state()
1950 def delete_vl_instance(self
, vld
):
1951 for vlr
in self
._vlrs
:
1952 if vlr
.vld_msg
.id == vld
.id:
1953 self
._log
.debug("Found VLR %s for VLD %s in NSR %s",
1954 vlr
.id, vld
.id, self
.id)
1955 vlr
.state
= VlRecordState
.TERMINATE_PENDING
1956 yield from self
.update_state()
1959 yield from self
.nsm_plugin
.terminate_vl(vlr
)
1960 vlr
.state
= VlRecordState
.TERMINATED
1961 self
._vlrs
.remove(vlr
)
1963 except Exception as e
:
1964 err_msg
= "Error terminating VL for NSR {} and VLD {}: {}". \
1965 format(self
.id, vld
.id, e
)
1966 self
._log
.error(err_msg
)
1967 self
._log
.exception(e
)
1968 vlr
.state
= VlRecordState
.FAILED
1970 yield from self
.update_state()
1974 def create_vnfs(self
, config_xact
):
1976 This function creates VNFs for every VNF in the NSD
1977 associated with this NSR
1979 self
._log
.debug("Creating %u VNFs associated with this NS id %s",
1980 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
1982 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1983 if not const_vnfd
.start_by_default
:
1984 self
._log
.debug("start_by_default set to False in constituent VNF (%s). Skipping start.",
1985 const_vnfd
.member_vnf_index
)
1988 vnfd_msg
= self
._get
_vnfd
(const_vnfd
.vnfd_id_ref
, config_xact
)
1989 cloud_account_name
,om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd
.member_vnf_index
)
1990 if cloud_account_name
is None:
1991 cloud_account_name
= self
.cloud_account_name
1992 yield from self
.create_vnf_record(vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
)
1995 def get_placement_groups(self
, vnfd_msg
, const_vnfd
):
1996 placement_groups
= []
1997 for group
in self
.nsd_msg
.placement_groups
:
1998 for member_vnfd
in group
.member_vnfd
:
1999 if (member_vnfd
.vnfd_id_ref
== vnfd_msg
.id) and \
2000 (member_vnfd
.member_vnf_index_ref
== const_vnfd
.member_vnf_index
):
2001 group_info
= self
.resolve_placement_group_cloud_construct(group
)
2002 if group_info
is None:
2003 self
._log
.error("Could not resolve cloud-construct for placement group: %s", group
.name
)
2004 ### raise PlacementGroupError("Could not resolve cloud-construct for placement group: {}".format(group.name))
2006 self
._log
.info("Successfully resolved cloud construct for placement group: %s for VNF: %s (Member Index: %s)",
2009 const_vnfd
.member_vnf_index
)
2010 placement_groups
.append(group_info
)
2011 return placement_groups
2014 def create_vnf_record(self
, vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
, group_name
=None, group_instance_id
=None):
2015 # Fetch the VNFD associated with this VNF
2016 placement_groups
= self
.get_placement_groups(vnfd_msg
, const_vnfd
)
2017 self
._log
.info("Cloud Account for VNF %d is %s",const_vnfd
.member_vnf_index
,cloud_account_name
)
2018 self
._log
.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s",
2020 const_vnfd
.member_vnf_index
,
2021 [ group
.name
for group
in placement_groups
])
2022 vnfr
= yield from VirtualNetworkFunctionRecord
.create_record(self
._dts
,
2035 restart_mode
=self
.restart_mode
,
2037 if vnfr
.id in self
._vnfrs
:
2038 err
= "VNF with VNFR id %s already in vnf list" % (vnfr
.id,)
2039 raise NetworkServiceRecordError(err
)
2041 self
._vnfrs
[vnfr
.id] = vnfr
2042 self
._nsm
.vnfrs
[vnfr
.id] = vnfr
2044 yield from vnfr
.set_config_status(NsrYang
.ConfigStates
.INIT
)
2046 self
._log
.debug("Added VNFR %s to NSM VNFR list with id %s",
2052 def create_param_pools(self
):
2053 for param_pool
in self
.nsd_msg
.parameter_pool
:
2054 self
._log
.debug("Found parameter pool %s in nsr id %s", param_pool
, self
.id)
2056 start_value
= param_pool
.range.start_value
2057 end_value
= param_pool
.range.end_value
2058 if end_value
< start_value
:
2059 raise NetworkServiceRecordError(
2060 "Parameter pool %s has invalid range (start: {}, end: {})".format(
2061 start_value
, end_value
2065 self
._param
_pools
[param_pool
.name
] = config_value_pool
.ParameterValuePool(
2068 range(start_value
, end_value
)
2072 def fetch_vnfr(self
, vnfr_path
):
2073 """ Fetch VNFR record """
2075 self
._log
.debug("Fetching VNFR with key %s while instantiating %s",
2077 res_iter
= yield from self
._dts
.query_read(vnfr_path
, rwdts
.XactFlag
.MERGE
)
2079 for ent
in res_iter
:
2080 res
= yield from ent
2086 def instantiate_vnfs(self
, vnfrs
):
2088 This function instantiates VNFs for every VNF in this Network Service
2090 self
._log
.debug("Instantiating %u VNFs in NS %s", len(vnfrs
), self
.id)
2092 self
._log
.debug("Instantiating VNF: %s in NS %s", vnf
, self
.id)
2093 yield from self
.nsm_plugin
.instantiate_vnf(self
, vnf
)
2096 def instantiate_vnffgs(self
):
2098 This function instantiates VNFFGs for every VNFFG in this Network Service
2100 self
._log
.debug("Instantiating %u VNFFGs in NS %s",
2101 len(self
.nsd_msg
.vnffgd
), self
.id)
2102 for _
, vnfr
in self
.vnfrs
.items():
2103 while vnfr
.state
in [VnfRecordState
.INSTANTIATION_PENDING
, VnfRecordState
.INIT
]:
2104 self
._log
.debug("Received vnfr state for vnfr %s is %s; retrying",vnfr
.name
,vnfr
.state
)
2105 yield from asyncio
.sleep(2, loop
=self
._loop
)
2106 if vnfr
.state
== VnfRecordState
.ACTIVE
:
2107 self
._log
.debug("Received vnfr state for vnfr %s is %s ",vnfr
.name
,vnfr
.state
)
2110 self
._log
.debug("Received vnfr state for vnfr %s is %s; failing vnffg creation",vnfr
.name
,vnfr
.state
)
2111 self
._vnffgr
_state
= VnffgRecordState
.FAILED
2114 self
._log
.info("Waiting for 90 seconds for VMs to come up")
2115 yield from asyncio
.sleep(90, loop
=self
._loop
)
2116 self
._log
.info("Starting VNFFG orchestration")
2117 for vnffg
in self
._vnffgrs
.values():
2118 self
._log
.debug("Instantiating VNFFG: %s in NS %s", vnffg
, self
.id)
2119 yield from vnffg
.instantiate()
2122 def instantiate_scaling_instances(self
, config_xact
):
2123 """ Instantiate any default scaling instances in this Network Service """
2124 for group
in self
._scaling
_groups
.values():
2125 for i
in range(group
.min_instance_count
):
2126 self
._log
.debug("Instantiating %s default scaling instance %s", group
, i
)
2127 yield from self
.create_scale_group_instance(
2128 group
.name
, i
, config_xact
, is_default
=True
2131 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2132 if group_msg
.scaling_group_name_ref
!= group
.name
:
2135 for instance
in group_msg
.instance
:
2136 self
._log
.debug("Reloading %s scaling instance %s", group_msg
, instance
.id)
2137 yield from self
.create_scale_group_instance(
2138 group
.name
, instance
.id, config_xact
, is_default
=False
2141 def has_scaling_instances(self
):
2142 """ Return boolean indicating if the network service has default scaling groups """
2143 for group
in self
._scaling
_groups
.values():
2144 if group
.min_instance_count
> 0:
2147 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2148 if len(group_msg
.instance
) > 0:
2155 """ This function publishes this NSR """
2156 self
._nsr
_msg
= self
.create_msg()
2158 self
._log
.debug("Publishing the NSR with xpath %s and nsr %s",
2162 if self
._debug
_running
:
2163 self
._log
.debug("Publishing NSR in RUNNING state!")
2166 with self
._dts
.transaction() as xact
:
2167 yield from self
._nsm
.nsr_handler
.update(xact
, self
.nsr_xpath
, self
._nsr
_msg
)
2168 if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
:
2169 self
._debug
_running
= True
2172 def unpublish(self
, xact
):
2173 """ Unpublish this NSR object """
2174 self
._log
.debug("Unpublishing Network service id %s", self
.id)
2175 yield from self
._nsm
.nsr_handler
.delete(xact
, self
.nsr_xpath
)
2178 def nsr_xpath(self
):
2179 """ Returns the xpath associated with this NSR """
2181 "D,/nsr:ns-instance-opdata" +
2182 "/nsr:nsr[nsr:ns-instance-config-ref = '{}']"
2186 def xpath_from_nsr(nsr
):
2187 """ Returns the xpath associated with this NSR op data"""
2188 return (NetworkServiceRecord
.XPATH
+
2189 "[nsr:ns-instance-config-ref = '{}']").format(nsr
.id)
2192 def nsd_xpath(self
):
2193 """ Return NSD config xpath."""
2195 "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}']"
2196 ).format(self
.nsd_id
)
2199 def instantiate(self
, config_xact
):
2200 """"Instantiates a NetworkServiceRecord.
2202 This function instantiates a Network service
2203 which involves the following steps,
2205 * Instantiate every VL in NSD by sending create VLR request to DTS.
2206 * Instantiate every VNF in NSD by sending create VNF reuqest to DTS.
2207 * Publish the NSR details to DTS
2210 nsr: The NSR configuration request containing nsr-id and nsd
2211 config_xact: The configuration transaction which initiated the instatiation
2214 NetworkServiceRecordError if the NSR creation fails
2220 self
._log
.debug("Instantiating NS - %s xact - %s", self
, config_xact
)
2222 # Move the state to INIITALIZING
2223 self
.set_state(NetworkServiceRecordState
.INIT
)
2225 event_descr
= "Instantiation Request Received NSR Id:%s" % self
.id
2226 self
.record_event("instantiating", event_descr
)
2229 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
2232 # Update ref count if nsd present in catalog
2233 self
._nsm
.get_nsd_ref(self
.nsd_id
)
2235 except NetworkServiceDescriptorError
:
2236 # This could be an NSD not in the nsd-catalog
2239 # Merge any config and initial config primitive values
2240 self
.config_store
.merge_nsd_config(self
.nsd_msg
)
2241 self
._log
.debug("Merged NSD: {}".format(self
.nsd_msg
.as_dict()))
2243 event_descr
= "Fetched NSD with descriptor id %s" % self
.nsd_id
2244 self
.record_event("nsd-fetched", event_descr
)
2246 if self
._nsd
is None:
2247 msg
= "Failed to fetch NSD with nsd-id [%s] for nsr-id %s"
2248 self
._log
.debug(msg
, self
.nsd_id
, self
.id)
2249 raise NetworkServiceRecordError(self
)
2251 self
._log
.debug("Got nsd result %s", self
._nsd
)
2253 # Substitute any input parameters
2254 self
.substitute_input_parameters(self
._nsd
, self
._nsr
_cfg
_msg
)
2257 yield from self
.create(config_xact
)
2259 # Publish the NSR to DTS
2260 yield from self
.publish()
2263 def do_instantiate():
2265 Instantiate network service
2267 self
._log
.debug("Instantiating VLs nsr id [%s] nsd id [%s]",
2268 self
.id, self
.nsd_id
)
2270 # instantiate the VLs
2271 event_descr
= ("Instantiating %s external VLs for NSR id %s" %
2272 (len(self
.nsd_msg
.vld
), self
.id))
2273 self
.record_event("begin-external-vls-instantiation", event_descr
)
2275 self
.set_state(NetworkServiceRecordState
.VL_INIT_PHASE
)
2277 yield from self
.instantiate_vls()
2279 # Publish the NSR to DTS
2280 yield from self
.publish()
2282 event_descr
= ("Finished instantiating %s external VLs for NSR id %s" %
2283 (len(self
.nsd_msg
.vld
), self
.id))
2284 self
.record_event("end-external-vls-instantiation", event_descr
)
2286 self
.set_state(NetworkServiceRecordState
.VNF_INIT_PHASE
)
2288 self
._log
.debug("Instantiating VNFs ...... nsr[%s], nsd[%s]",
2289 self
.id, self
.nsd_id
)
2291 # instantiate the VNFs
2292 event_descr
= ("Instantiating %s VNFS for NSR id %s" %
2293 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2295 self
.record_event("begin-vnf-instantiation", event_descr
)
2297 yield from self
.instantiate_vnfs(self
._vnfrs
.values())
2299 self
._log
.debug(" Finished instantiating %d VNFs for NSR id %s",
2300 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
2302 event_descr
= ("Finished instantiating %s VNFs for NSR id %s" %
2303 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2304 self
.record_event("end-vnf-instantiation", event_descr
)
2306 if len(self
.vnffgrs
) > 0:
2307 #self.set_state(NetworkServiceRecordState.VNFFG_INIT_PHASE)
2308 event_descr
= ("Instantiating %s VNFFGS for NSR id %s" %
2309 (len(self
.nsd_msg
.vnffgd
), self
.id))
2311 self
.record_event("begin-vnffg-instantiation", event_descr
)
2313 yield from self
.instantiate_vnffgs()
2315 event_descr
= ("Finished instantiating %s VNFFGDs for NSR id %s" %
2316 (len(self
.nsd_msg
.vnffgd
), self
.id))
2317 self
.record_event("end-vnffg-instantiation", event_descr
)
2319 if self
.has_scaling_instances():
2320 event_descr
= ("Instantiating %s Scaling Groups for NSR id %s" %
2321 (len(self
._scaling
_groups
), self
.id))
2323 self
.record_event("begin-scaling-group-instantiation", event_descr
)
2324 yield from self
.instantiate_scaling_instances(config_xact
)
2325 self
.record_event("end-scaling-group-instantiation", event_descr
)
2327 # Give the plugin a chance to deploy the network service now that all
2328 # virtual links and vnfs are instantiated
2329 yield from self
.nsm_plugin
.deploy(self
._nsr
_msg
)
2331 self
._log
.debug("Publishing NSR...... nsr[%s], nsd[%s]",
2332 self
.id, self
.nsd_id
)
2334 # Publish the NSR to DTS
2335 yield from self
.publish()
2337 self
._log
.debug("Published NSR...... nsr[%s], nsd[%s]",
2338 self
.id, self
.nsd_id
)
2340 def on_instantiate_done(fut
):
2341 # If the do_instantiate fails, then publish NSR with failed result
2342 if fut
.exception() is not None:
2343 self
._log
.error("NSR instantiation failed for NSR id %s: %s", self
.id, str(fut
.exception()))
2344 self
._loop
.create_task(self
.instantiation_failed(failed_reason
=str(fut
.exception())))
2346 instantiate_task
= self
._loop
.create_task(do_instantiate())
2347 instantiate_task
.add_done_callback(on_instantiate_done
)
2350 def set_config_status(self
, status
, status_details
=None):
2351 if self
.config_status
!= status
:
2352 self
._log
.debug("Updating NSR {} status for {} to {}".
2353 format(self
.name
, self
.config_status
, status
))
2354 self
._config
_status
= status
2355 self
._config
_status
_details
= status_details
2357 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2358 self
.record_event("config-failed", "NS configuration failed",
2359 evt_details
=self
._config
_status
_details
)
2361 yield from self
.publish()
2364 def is_active(self
):
2365 """ This NS is active """
2366 self
.set_state(NetworkServiceRecordState
.RUNNING
)
2370 # Publish the NSR to DTS
2371 self
._log
.debug("Network service %s is active ", self
.id)
2372 self
._is
_active
= True
2374 event_descr
= "NSR in running state for NSR id %s" % self
.id
2375 self
.record_event("ns-running", event_descr
)
2377 yield from self
.publish()
2380 def instantiation_failed(self
, failed_reason
=None):
2381 """ The NS instantiation failed"""
2382 self
._log
.error("Network service id:%s, name:%s instantiation failed",
2384 self
.set_state(NetworkServiceRecordState
.FAILED
)
2386 event_descr
= "Instantiation of NS %s failed" % self
.id
2387 self
.record_event("ns-failed", event_descr
, evt_details
=failed_reason
)
2389 # Publish the NSR to DTS
2390 yield from self
.publish()
2393 def terminate_vnfrs(self
, vnfrs
):
2394 """ Terminate VNFRS in this network service """
2395 self
._log
.debug("Terminating VNFs in network service %s", self
.id)
2397 yield from self
.nsm_plugin
.terminate_vnf(vnfr
)
2400 def terminate(self
):
2401 """ Terminate a NetworkServiceRecord."""
2402 def terminate_vnffgrs():
2403 """ Terminate VNFFGRS in this network service """
2404 self
._log
.debug("Terminating VNFFGRs in network service %s", self
.id)
2405 for vnffgr
in self
.vnffgrs
.values():
2406 yield from vnffgr
.terminate()
2408 def terminate_vlrs():
2409 """ Terminate VLRs in this netork service """
2410 self
._log
.debug("Terminating VLs in network service %s", self
.id)
2411 for vlr
in self
.vlrs
:
2412 yield from self
.nsm_plugin
.terminate_vl(vlr
)
2413 vlr
.state
= VlRecordState
.TERMINATED
2415 self
._log
.debug("Terminating network service id %s", self
.id)
2417 # Move the state to TERMINATE
2418 self
.set_state(NetworkServiceRecordState
.TERMINATE
)
2419 event_descr
= "Terminate being processed for NS Id:%s" % self
.id
2420 self
.record_event("terminate", event_descr
)
2422 # Move the state to VNF_TERMINATE_PHASE
2423 self
._log
.debug("Terminating VNFFGs in NS ID: %s", self
.id)
2424 self
.set_state(NetworkServiceRecordState
.VNFFG_TERMINATE_PHASE
)
2425 event_descr
= "Terminating VNFFGS in NS Id:%s" % self
.id
2426 self
.record_event("terminating-vnffgss", event_descr
)
2427 yield from terminate_vnffgrs()
2429 # Move the state to VNF_TERMINATE_PHASE
2430 self
.set_state(NetworkServiceRecordState
.VNF_TERMINATE_PHASE
)
2431 event_descr
= "Terminating VNFS in NS Id:%s" % self
.id
2432 self
.record_event("terminating-vnfs", event_descr
)
2433 yield from self
.terminate_vnfrs(self
.vnfrs
.values())
2435 # Move the state to VL_TERMINATE_PHASE
2436 self
.set_state(NetworkServiceRecordState
.VL_TERMINATE_PHASE
)
2437 event_descr
= "Terminating VLs in NS Id:%s" % self
.id
2438 self
.record_event("terminating-vls", event_descr
)
2439 yield from terminate_vlrs()
2441 yield from self
.nsm_plugin
.terminate_ns(self
)
2443 # Move the state to TERMINATED
2444 self
.set_state(NetworkServiceRecordState
.TERMINATED
)
2445 event_descr
= "Terminated NS Id:%s" % self
.id
2446 self
.record_event("terminated", event_descr
)
2449 """"Enable a NetworkServiceRecord."""
2453 """"Disable a NetworkServiceRecord."""
2456 def map_config_status(self
):
2457 self
._log
.debug("Config status for ns {} is {}".
2458 format(self
.name
, self
._config
_status
))
2459 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURING
:
2460 return 'configuring'
2461 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2465 def vl_phase_completed(self
):
2466 """ Are VLs created in this NS?"""
2467 return self
._vl
_phase
_completed
2469 def vnf_phase_completed(self
):
2470 """ Are VLs created in this NS?"""
2471 return self
._vnf
_phase
_completed
2473 def create_msg(self
):
2474 """ The network serice record as a message """
2475 nsr_dict
= {"ns_instance_config_ref": self
.id}
2476 nsr
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr
.from_dict(nsr_dict
)
2477 #nsr.cloud_account = self.cloud_account_name
2478 nsr
.sdn_account
= self
._sdn
_account
_name
2479 nsr
.name_ref
= self
.name
2480 nsr
.nsd_ref
= self
.nsd_id
2481 nsr
.nsd_name_ref
= self
.nsd_msg
.name
2482 nsr
.operational_events
= self
._op
_status
.msg
2483 nsr
.operational_status
= self
._op
_status
.yang_str()
2484 nsr
.config_status
= self
.map_config_status()
2485 nsr
.config_status_details
= self
._config
_status
_details
2486 nsr
.create_time
= self
._create
_time
2487 nsr
.uptime
= int(time
.time()) - self
._create
_time
2489 for cfg_prim
in self
.nsd_msg
.service_primitive
:
2490 cfg_prim
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_ServicePrimitive
.from_dict(
2492 nsr
.service_primitive
.append(cfg_prim
)
2494 for init_cfg
in self
.nsd_msg
.initial_config_primitive
:
2495 prim
= NsrYang
.NsrInitialConfigPrimitive
.from_dict(
2497 nsr
.initial_config_primitive
.append(prim
)
2499 if self
.vl_phase_completed():
2500 for vlr
in self
.vlrs
:
2501 nsr
.vlr
.append(vlr
.create_nsr_vlr_msg(self
.vnfrs
.values()))
2503 if self
.vnf_phase_completed():
2504 for vnfr_id
in self
.vnfrs
:
2505 nsr
.constituent_vnfr_ref
.append(self
.vnfrs
[vnfr_id
].const_vnfr_msg
)
2506 for vnffgr
in self
.vnffgrs
.values():
2507 nsr
.vnffgr
.append(vnffgr
.fetch_vnffgr())
2508 for scaling_group
in self
._scaling
_groups
.values():
2509 nsr
.scaling_group_record
.append(scaling_group
.create_record_msg())
2513 def all_vnfs_active(self
):
2514 """ Are all VNFS in this NS active? """
2515 for _
, vnfr
in self
.vnfrs
.items():
2516 if vnfr
.active
is not True:
2521 def update_state(self
):
2522 """ Re-evaluate this NS's state """
2523 curr_state
= self
._op
_status
.state
2525 if curr_state
== NetworkServiceRecordState
.TERMINATED
:
2526 self
._log
.debug("NS (%s) in terminated state, not updating state", self
.id)
2529 new_state
= NetworkServiceRecordState
.RUNNING
2530 self
._log
.info("Received update_state for nsr: %s, curr-state: %s",
2531 self
.id, curr_state
)
2533 # Check all the VNFRs are present
2534 for _
, vnfr
in self
.vnfrs
.items():
2535 if vnfr
.state
in [VnfRecordState
.ACTIVE
, VnfRecordState
.TERMINATED
]:
2537 elif vnfr
.state
== VnfRecordState
.FAILED
:
2538 if vnfr
._prev
_state
!= vnfr
.state
:
2539 event_descr
= "Instantiation of VNF %s failed" % vnfr
.id
2540 event_error_details
= vnfr
.state_failed_reason
2541 self
.record_event("vnf-failed", event_descr
, evt_details
=event_error_details
)
2542 vnfr
.set_state(VnfRecordState
.FAILED
)
2544 self
._log
.info("VNF state did not change, curr=%s, prev=%s",
2545 vnfr
.state
, vnfr
._prev
_state
)
2546 new_state
= NetworkServiceRecordState
.FAILED
2549 self
._log
.info("VNF %s in NSR %s is still not active; current state is: %s",
2550 vnfr
.id, self
.id, vnfr
.state
)
2551 new_state
= curr_state
2553 # If new state is RUNNING; check all VLs
2554 if new_state
== NetworkServiceRecordState
.RUNNING
:
2555 for vl
in self
.vlrs
:
2557 if vl
.state
in [VlRecordState
.ACTIVE
, VlRecordState
.TERMINATED
]:
2559 elif vl
.state
== VlRecordState
.FAILED
:
2560 if vl
.prev_state
!= vl
.state
:
2561 event_descr
= "Instantiation of VL %s failed" % vl
.id
2562 event_error_details
= vl
.state_failed_reason
2563 self
.record_event("vl-failed", event_descr
, evt_details
=event_error_details
)
2564 vl
.prev_state
= vl
.state
2566 self
._log
.debug("VL %s already in failed state")
2568 if vl
.state
in [VlRecordState
.INSTANTIATION_PENDING
, VlRecordState
.INIT
]:
2569 new_state
= NetworkServiceRecordState
.VL_INSTANTIATE
2572 if vl
.state
in [VlRecordState
.TERMINATE_PENDING
]:
2573 new_state
= NetworkServiceRecordState
.VL_TERMINATE
2576 # If new state is RUNNING; check VNFFGRs are also active
2577 if new_state
== NetworkServiceRecordState
.RUNNING
:
2578 for _
, vnffgr
in self
.vnffgrs
.items():
2579 self
._log
.info("Checking vnffgr state for nsr %s is: %s",
2580 self
.id, vnffgr
.state
)
2581 if vnffgr
.state
== VnffgRecordState
.ACTIVE
:
2583 elif vnffgr
.state
== VnffgRecordState
.FAILED
:
2584 event_descr
= "Instantiation of VNFFGR %s failed" % vnffgr
.id
2585 self
.record_event("vnffg-failed", event_descr
)
2586 new_state
= NetworkServiceRecordState
.FAILED
2589 self
._log
.info("VNFFGR %s in NSR %s is still not active; current state is: %s",
2590 vnffgr
.id, self
.id, vnffgr
.state
)
2591 new_state
= curr_state
2593 # Update all the scaling group instance operational status to
2594 # reflect the state of all VNFR within that instance
2595 yield from self
._update
_scale
_group
_instances
_status
()
2597 for _
, group
in self
._scaling
_groups
.items():
2598 if group
.state
== scale_group
.ScaleGroupState
.SCALING_OUT
:
2599 new_state
= NetworkServiceRecordState
.SCALING_OUT
2601 elif group
.state
== scale_group
.ScaleGroupState
.SCALING_IN
:
2602 new_state
= NetworkServiceRecordState
.SCALING_IN
2605 if new_state
!= curr_state
:
2606 self
._log
.debug("Changing state of Network service %s from %s to %s",
2607 self
.id, curr_state
, new_state
)
2608 if new_state
== NetworkServiceRecordState
.RUNNING
:
2609 yield from self
.is_active()
2610 elif new_state
== NetworkServiceRecordState
.FAILED
:
2611 # If the NS is already active and we entered scaling_in, scaling_out,
2612 # do not mark the NS as failing if scaling operation failed.
2613 if curr_state
in [NetworkServiceRecordState
.SCALING_OUT
,
2614 NetworkServiceRecordState
.SCALING_IN
] and self
._is
_active
:
2615 new_state
= NetworkServiceRecordState
.RUNNING
2616 self
.set_state(new_state
)
2618 yield from self
.instantiation_failed()
2620 self
.set_state(new_state
)
2622 yield from self
.publish()
2625 class InputParameterSubstitution(object):
2627 This class is responsible for substituting input parameters into an NSD.
2630 def __init__(self
, log
):
2631 """Create an instance of InputParameterSubstitution
2634 log - a logger for this object to use
2639 def __call__(self
, nsd
, nsr_config
):
2640 """Substitutes input parameters from the NSR config into the NSD
2642 This call modifies the provided NSD with the input parameters that are
2643 contained in the NSR config.
2646 nsd - a GI NSD object
2647 nsr_config - a GI NSR config object
2650 if nsd
is None or nsr_config
is None:
2653 # Create a lookup of the xpath elements that this descriptor allows
2655 optional_input_parameters
= set()
2656 for input_parameter
in nsd
.input_parameter_xpath
:
2657 optional_input_parameters
.add(input_parameter
.xpath
)
2659 # Apply the input parameters to the descriptor
2660 if nsr_config
.input_parameter
:
2661 for param
in nsr_config
.input_parameter
:
2662 if param
.xpath
not in optional_input_parameters
:
2663 msg
= "tried to set an invalid input parameter ({})"
2664 self
.log
.error(msg
.format(param
.xpath
))
2668 "input-parameter:{} = {}".format(
2675 xpath
.setxattr(nsd
, param
.xpath
, param
.value
)
2677 except Exception as e
:
2678 self
.log
.exception(e
)
2681 class NetworkServiceDescriptor(object):
2683 Network service descriptor class
2686 def __init__(self
, dts
, log
, loop
, nsd
, nsm
):
2698 """ Returns nsd id """
2703 """ Returns name of nsd """
2704 return self
._nsd
.name
2707 def ref_count(self
):
2708 """ Returns reference count"""
2709 return self
._ref
_count
2712 """ Returns whether nsd is in use or not """
2713 return True if self
.ref_count
> 0 else False
2716 """ Take a reference on this object """
2717 self
._ref
_count
+= 1
2720 """ Release reference on this object """
2721 if self
.ref_count
< 1:
2722 msg
= ("Unref on a NSD object - nsd id %s, ref_count = %s" %
2723 (self
.id, self
.ref_count
))
2724 self
._log
.critical(msg
)
2725 raise NetworkServiceDescriptorError(msg
)
2726 self
._ref
_count
-= 1
2730 """ Return the message associated with this NetworkServiceDescriptor"""
2734 def path_for_id(nsd_id
):
2735 """ Return path for the passed nsd_id"""
2736 return "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}'".format(nsd_id
)
2739 """ Return the message associated with this NetworkServiceDescriptor"""
2740 return NetworkServiceDescriptor
.path_for_id(self
.id)
2742 def update(self
, nsd
):
2743 """ Update the NSD descriptor """
2747 class NsdDtsHandler(object):
2748 """ The network service descriptor DTS handler """
2749 XPATH
= "C,/nsd:nsd-catalog/nsd:nsd"
2751 def __init__(self
, dts
, log
, loop
, nsm
):
2761 """ Return registration handle """
2766 """ Register for Nsd create/update/delete/read requests from dts """
2768 def on_apply(dts
, acg
, xact
, action
, scratch
):
2769 """Apply the configuration"""
2770 is_recovery
= xact
.xact
is None and action
== rwdts
.AppconfAction
.INSTALL
2771 self
._log
.debug("Got nsd apply cfg (xact:%s) (action:%s)",
2773 # Create/Update an NSD record
2774 for cfg
in self
._regh
.get_xact_elements(xact
):
2775 # Only interested in those NSD cfgs whose ID was received in prepare callback
2776 if cfg
.id in scratch
.get('nsds', []) or is_recovery
:
2777 self
._nsm
.update_nsd(cfg
)
2779 scratch
.pop('nsds', None)
2781 return RwTypes
.RwStatus
.SUCCESS
2784 def delete_nsd_libs(nsd_id
):
2785 """ Remove any files uploaded with NSD and stored under $RIFT_ARTIFACTS/libs/<id> """
2787 rift_artifacts_dir
= os
.environ
['RIFT_ARTIFACTS']
2788 nsd_dir
= os
.path
.join(rift_artifacts_dir
, 'launchpad/libs', nsd_id
)
2790 if os
.path
.exists (nsd_dir
):
2791 shutil
.rmtree(nsd_dir
, ignore_errors
=True)
2792 except Exception as e
:
2793 self
._log
.error("Exception in cleaning up NSD libs {}: {}".
2795 self
._log
.excpetion(e
)
2798 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2799 """ Prepare callback from DTS for NSD config """
2801 self
._log
.info("Got nsd prepare - config received nsd id %s, msg %s",
2804 fref
= ProtobufC
.FieldReference
.alloc()
2805 fref
.goto_whole_message(msg
.to_pbcm())
2807 if fref
.is_field_deleted():
2808 # Delete an NSD record
2809 self
._log
.debug("Deleting NSD with id %s", msg
.id)
2810 if self
._nsm
.nsd_in_use(msg
.id):
2811 self
._log
.debug("Cannot delete NSD in use - %s", msg
.id)
2812 err
= "Cannot delete an NSD in use - %s" % msg
.id
2813 raise NetworkServiceDescriptorRefCountExists(err
)
2815 yield from delete_nsd_libs(msg
.id)
2816 self
._nsm
.delete_nsd(msg
.id)
2818 # Add this NSD to scratch to create/update in apply callback
2819 nsds
= scratch
.setdefault('nsds', [])
2821 # acg._scratch['nsds'].append(msg.id)
2823 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2826 "Registering for NSD config using xpath: %s",
2827 NsdDtsHandler
.XPATH
,
2830 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2831 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2832 # Need a list in scratch to store NSDs to create/update later
2833 # acg._scratch['nsds'] = list()
2834 self
._regh
= acg
.register(
2835 xpath
=NsdDtsHandler
.XPATH
,
2836 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
2837 on_prepare
=on_prepare
)
2840 class VnfdDtsHandler(object):
2841 """ DTS handler for VNFD config changes """
2842 XPATH
= "C,/vnfd:vnfd-catalog/vnfd:vnfd"
2844 def __init__(self
, dts
, log
, loop
, nsm
):
2853 """ DTS registration handle """
2858 """ Register for VNFD configuration"""
2861 def on_apply(dts
, acg
, xact
, action
, scratch
):
2862 """Apply the configuration"""
2863 self
._log
.debug("Got NSM VNFD apply (xact: %s) (action: %s)(scr: %s)",
2864 xact
, action
, scratch
)
2866 # Create/Update a VNFD record
2867 for cfg
in self
._regh
.get_xact_elements(xact
):
2868 # Only interested in those VNFD cfgs whose ID was received in prepare callback
2869 if cfg
.id in scratch
.get('vnfds', []):
2870 self
._nsm
.update_vnfd(cfg
)
2872 for cfg
in self
._regh
.elements
:
2873 if cfg
.id in scratch
.get('deleted_vnfds', []):
2874 yield from self
._nsm
.delete_vnfd(cfg
.id)
2876 scratch
.pop('vnfds', None)
2877 scratch
.pop('deleted_vnfds', None)
2880 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2881 """ on prepare callback """
2882 self
._log
.debug("Got on prepare for VNFD (path: %s) (action: %s) (msg: %s)",
2883 ks_path
.to_xpath(RwNsmYang
.get_schema()), xact_info
.query_action
, msg
)
2885 fref
= ProtobufC
.FieldReference
.alloc()
2886 fref
.goto_whole_message(msg
.to_pbcm())
2888 # Handle deletes in prepare_callback, but adds/updates in apply_callback
2889 if fref
.is_field_deleted():
2890 self
._log
.debug("Adding msg to deleted field")
2891 deleted_vnfds
= scratch
.setdefault('deleted_vnfds', [])
2892 deleted_vnfds
.append(msg
.id)
2894 # Add this VNFD to scratch to create/update in apply callback
2895 vnfds
= scratch
.setdefault('vnfds', [])
2896 vnfds
.append(msg
.id)
2898 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2901 "Registering for VNFD config using xpath: %s",
2902 VnfdDtsHandler
.XPATH
,
2904 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2905 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2906 # Need a list in scratch to store VNFDs to create/update later
2907 # acg._scratch['vnfds'] = list()
2908 # acg._scratch['deleted_vnfds'] = list()
2909 self
._regh
= acg
.register(
2910 xpath
=VnfdDtsHandler
.XPATH
,
2911 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY
,
2912 on_prepare
=on_prepare
)
2914 class NsrRpcDtsHandler(object):
2915 """ The network service instantiation RPC DTS handler """
2916 EXEC_NSR_CONF_XPATH
= "I,/nsr:start-network-service"
2917 EXEC_NSR_CONF_O_XPATH
= "O,/nsr:start-network-service"
2918 NETCONF_IP_ADDRESS
= "127.0.0.1"
2920 RESTCONF_PORT
= 8888
2921 NETCONF_USER
= "admin"
2922 NETCONF_PW
= "admin"
2923 REST_BASE_V2_URL
= 'https://{}:{}/v2/api/'.format("127.0.0.1",8888)
2925 def __init__(self
, dts
, log
, loop
, nsm
):
2932 self
._ns
_regh
= None
2934 self
._manager
= None
2935 self
._nsr
_config
_url
= NsrRpcDtsHandler
.REST_BASE_V2_URL
+ 'config/ns-instance-config'
2937 self
._model
= RwYang
.Model
.create_libncx()
2938 self
._model
.load_schema_ypbc(RwNsrYang
.get_schema())
2942 """ Return the NS manager instance """
2946 def wrap_netconf_config_xml(xml
):
2947 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
2951 def _connect(self
, timeout_secs
=240):
2953 start_time
= time
.time()
2954 while (time
.time() - start_time
) < timeout_secs
:
2957 self
._log
.debug("Attemping NsmTasklet netconf connection.")
2959 manager
= yield from ncclient
.asyncio_manager
.asyncio_connect(
2961 host
=NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
,
2962 port
=NsrRpcDtsHandler
.NETCONF_PORT
,
2963 username
=NsrRpcDtsHandler
.NETCONF_USER
,
2964 password
=NsrRpcDtsHandler
.NETCONF_PW
,
2966 look_for_keys
=False,
2967 hostkey_verify
=False,
2972 except ncclient
.transport
.errors
.SSHError
as e
:
2973 self
._log
.warning("Netconf connection to launchpad %s failed: %s",
2974 NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
, str(e
))
2976 yield from asyncio
.sleep(5, loop
=self
._loop
)
2978 raise NsrInstantiationFailed("Failed to connect to Launchpad within %s seconds" %
2981 def _apply_ns_instance_config(self
,payload_dict
):
2982 #self._log.debug("At apply NS instance config with payload %s",payload_dict)
2983 req_hdr
= {'accept':'application/vnd.yang.data+json','content-type':'application/vnd.yang.data+json'}
2984 response
=requests
.post(self
._nsr
_config
_url
, headers
=req_hdr
, auth
=('admin', 'admin'),data
=payload_dict
,verify
=False)
2989 """ Register for NS monitoring read from dts """
2991 def on_ns_config_prepare(xact_info
, action
, ks_path
, msg
):
2992 """ prepare callback from dts start-network-service"""
2993 assert action
== rwdts
.QueryAction
.RPC
2995 rpc_op
= NsrYang
.YangOutput_Nsr_StartNetworkService
.from_dict({
2996 "nsr_id":str(uuid
.uuid4())
2999 if not ('name' in rpc_ip
and 'nsd_ref' in rpc_ip
and ('cloud_account' in rpc_ip
or 'om_datacenter' in rpc_ip
)):
3000 self
._log
.error("Mandatory parameters name or nsd_ref or cloud account not found in start-network-service {}".format(rpc_ip
))
3003 self
._log
.debug("start-network-service RPC input: {}".format(rpc_ip
))
3006 # Add used value to the pool
3007 self
._log
.debug("RPC output: {}".format(rpc_op
))
3009 nsd_copy
= self
.nsm
.get_nsd(rpc_ip
.nsd_ref
)
3011 #if not self._manager:
3012 # self._manager = yield from self._connect()
3014 self
._log
.debug("Configuring ns-instance-config with name %s nsd-ref: %s",
3015 rpc_ip
.name
, rpc_ip
.nsd_ref
)
3017 ns_instance_config_dict
= {"id":rpc_op
.nsr_id
, "admin_status":"ENABLED"}
3018 ns_instance_config_copy_dict
= {k
:v
for k
, v
in rpc_ip
.as_dict().items()
3019 if k
in RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr().fields
}
3020 ns_instance_config_dict
.update(ns_instance_config_copy_dict
)
3022 ns_instance_config
= RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.from_dict(ns_instance_config_dict
)
3023 ns_instance_config
.nsd
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_Nsd()
3024 ns_instance_config
.nsd
.from_dict(nsd_copy
.msg
.as_dict())
3026 payload_dict
= ns_instance_config
.to_json(self
._model
)
3027 #xml = ns_instance_config.to_xml_v2(self._model)
3028 #netconf_xml = self.wrap_netconf_config_xml(xml)
3030 #self._log.debug("Sending configure ns-instance-config xml to %s: %s",
3031 # netconf_xml, NsrRpcDtsHandler.NETCONF_IP_ADDRESS)
3032 self
._log
.debug("Sending configure ns-instance-config json to %s: %s",
3033 self
._nsr
_config
_url
,ns_instance_config
)
3035 #response = yield from self._manager.edit_config(
3037 # config=netconf_xml,
3039 response
= yield from self
._loop
.run_in_executor(
3041 self
._apply
_ns
_instance
_config
,
3044 response
.raise_for_status()
3045 self
._log
.debug("Received edit config response: %s", response
.json())
3047 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
,
3048 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
,
3050 except Exception as e
:
3051 self
._log
.error("Exception processing the "
3052 "start-network-service: {}".format(e
))
3053 self
._log
.exception(e
)
3054 xact_info
.respond_xpath(rwdts
.XactRspCode
.NACK
,
3055 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
)
3058 hdl_ns
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_ns_config_prepare
,)
3060 with self
._dts
.group_create() as group
:
3061 self
._ns
_regh
= group
.register(xpath
=NsrRpcDtsHandler
.EXEC_NSR_CONF_XPATH
,
3063 flags
=rwdts
.Flag
.PUBLISHER
,
3067 class NsrDtsHandler(object):
3068 """ The network service DTS handler """
3069 NSR_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr"
3070 SCALE_INSTANCE_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr/nsr:scaling-group/nsr:instance"
3071 KEY_PAIR_XPATH
= "C,/nsr:key-pair"
3073 def __init__(self
, dts
, log
, loop
, nsm
):
3079 self
._nsr
_regh
= None
3080 self
._scale
_regh
= None
3081 self
._key
_pair
_regh
= None
3085 """ Return the NS manager instance """
3090 """ Register for Nsr create/update/delete/read requests from dts """
3092 def nsr_id_from_keyspec(ks
):
3093 nsr_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.schema().keyspec_to_entry(ks
)
3094 nsr_id
= nsr_path_entry
.key00
.id
3097 def group_name_from_keyspec(ks
):
3098 group_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
.schema().keyspec_to_entry(ks
)
3099 group_name
= group_path_entry
.key00
.scaling_group_name_ref
3102 def is_instance_in_reg_elements(nsr_id
, group_name
, instance_id
):
3103 """ Return boolean indicating if scaling group instance was already commited previously.
3105 By looking at the existing elements in this registration handle (elements not part
3106 of this current xact), we can tell if the instance was configured previously without
3107 keeping any application state.
3109 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3110 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3111 elem_group_name
= group_name_from_keyspec(keyspec
)
3113 if elem_nsr_id
!= nsr_id
or group_name
!= elem_group_name
:
3116 if instance_cfg
.id == instance_id
:
3121 def get_scale_group_instance_delta(nsr_id
, group_name
, xact
):
3122 delta
= {"added": [], "deleted": []}
3123 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3124 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3125 if elem_nsr_id
!= nsr_id
:
3128 elem_group_name
= group_name_from_keyspec(keyspec
)
3129 if elem_group_name
!= group_name
:
3132 delta
["added"].append(instance_cfg
.id)
3134 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(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 if instance_cfg
.id in delta
["added"]:
3144 delta
["added"].remove(instance_cfg
.id)
3146 delta
["deleted"].append(instance_cfg
.id)
3151 def update_nsr_nsd(nsr_id
, xact
, scratch
):
3154 def get_nsr_vl_delta(nsr_id
, xact
, scratch
):
3155 delta
= {"added": [], "deleted": []}
3156 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3157 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3158 if elem_nsr_id
!= nsr_id
:
3161 if 'vld' in instance_cfg
.nsd
:
3162 for vld
in instance_cfg
.nsd
.vld
:
3163 delta
["added"].append(vld
)
3165 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3166 self
._log
.debug("NSR update: %s", instance_cfg
)
3167 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3168 if elem_nsr_id
!= nsr_id
:
3171 if 'vld' in instance_cfg
.nsd
:
3172 for vld
in instance_cfg
.nsd
.vld
:
3173 if vld
in delta
["added"]:
3174 delta
["added"].remove(vld
)
3176 delta
["deleted"].append(vld
)
3180 vl_delta
= yield from get_nsr_vl_delta(nsr_id
, xact
, scratch
)
3181 self
._log
.debug("Got NSR:%s VL instance delta: %s", nsr_id
, vl_delta
)
3183 for vld
in vl_delta
["added"]:
3184 yield from self
._nsm
.nsr_instantiate_vl(nsr_id
, vld
)
3186 for vld
in vl_delta
["deleted"]:
3187 yield from self
._nsm
.nsr_terminate_vl(nsr_id
, vld
)
3189 def get_add_delete_update_cfgs(dts_member_reg
, xact
, key_name
, scratch
):
3190 # Unfortunately, it is currently difficult to figure out what has exactly
3191 # changed in this xact without Pbdelta support (RIFT-4916)
3192 # As a workaround, we can fetch the pre and post xact elements and
3193 # perform a comparison to figure out adds/deletes/updates
3194 xact_cfgs
= list(dts_member_reg
.get_xact_elements(xact
))
3195 curr_cfgs
= list(dts_member_reg
.elements
)
3197 xact_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in xact_cfgs
}
3198 curr_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in curr_cfgs
}
3201 added_keys
= set(xact_key_map
) - set(curr_key_map
)
3202 added_cfgs
= [xact_key_map
[key
] for key
in added_keys
]
3205 deleted_keys
= set(curr_key_map
) - set(xact_key_map
)
3206 deleted_cfgs
= [curr_key_map
[key
] for key
in deleted_keys
]
3209 updated_keys
= set(curr_key_map
) & set(xact_key_map
)
3210 updated_cfgs
= [xact_key_map
[key
] for key
in updated_keys
3211 if xact_key_map
[key
] != curr_key_map
[key
]]
3213 return added_cfgs
, deleted_cfgs
, updated_cfgs
3215 def get_nsr_key_pairs(dts_member_reg
, xact
):
3217 for instance_cfg
, keyspec
in dts_member_reg
.get_xact_elements(xact
, include_keyspec
=True):
3218 self
._log
.debug("Key pair received is {} KS: {}".format(instance_cfg
, keyspec
))
3219 xpath
= keyspec
.to_xpath(RwNsrYang
.get_schema())
3220 key_pairs
[instance_cfg
.name
] = instance_cfg
3223 def on_apply(dts
, acg
, xact
, action
, scratch
):
3224 """Apply the configuration"""
3225 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3226 xact
, action
, scratch
)
3228 def handle_create_nsr(msg
, key_pairs
=None, restart_mode
=False):
3229 # Handle create nsr requests """
3230 # Do some validations
3231 if not msg
.has_field("nsd"):
3232 err
= "NSD not provided"
3233 self
._log
.error(err
)
3234 raise NetworkServiceRecordError(err
)
3236 self
._log
.debug("Creating NetworkServiceRecord %s from nsr config %s",
3237 msg
.id, msg
.as_dict())
3238 nsr
= self
.nsm
.create_nsr(msg
, key_pairs
=key_pairs
, restart_mode
=restart_mode
)
3241 def handle_delete_nsr(msg
):
3243 def delete_instantiation(ns_id
):
3244 """ Delete instantiation """
3245 with self
._dts
.transaction() as xact
:
3246 yield from self
._nsm
.terminate_ns(ns_id
, xact
)
3248 # Handle delete NSR requests
3249 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3250 # Terminate the NSR instance
3251 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3253 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3254 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3255 nsr
.record_event("terminate-rcvd", event_descr
)
3257 self
._loop
.create_task(delete_instantiation(msg
.id))
3260 def begin_instantiation(nsr
):
3261 # Begin instantiation
3262 self
._log
.info("Beginning NS instantiation: %s", nsr
.id)
3263 yield from self
._nsm
.instantiate_ns(nsr
.id, xact
)
3265 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3266 xact
, action
, scratch
)
3268 if action
== rwdts
.AppconfAction
.INSTALL
and xact
.id is None:
3270 for element
in self
._key
_pair
_regh
.elements
:
3271 key_pairs
.append(element
)
3272 for element
in self
._nsr
_regh
.elements
:
3273 nsr
= handle_create_nsr(element
, key_pairs
, restart_mode
=True)
3274 self
._loop
.create_task(begin_instantiation(nsr
))
3277 (added_msgs
, deleted_msgs
, updated_msgs
) = get_add_delete_update_cfgs(self
._nsr
_regh
,
3281 self
._log
.debug("Added: %s, Deleted: %s, Updated: %s", added_msgs
,
3282 deleted_msgs
, updated_msgs
)
3284 for msg
in added_msgs
:
3285 if msg
.id not in self
._nsm
.nsrs
:
3286 self
._log
.info("Create NSR received in on_apply to instantiate NS:%s", msg
.id)
3287 key_pairs
= get_nsr_key_pairs(self
._key
_pair
_regh
, xact
)
3288 nsr
= handle_create_nsr(msg
,key_pairs
)
3289 self
._loop
.create_task(begin_instantiation(nsr
))
3291 for msg
in deleted_msgs
:
3292 self
._log
.info("Delete NSR received in on_apply to terminate NS:%s", msg
.id)
3294 handle_delete_nsr(msg
)
3296 self
._log
.exception("Failed to terminate NS:%s", msg
.id)
3298 for msg
in updated_msgs
:
3299 self
._log
.info("Update NSR received in on_apply: %s", msg
)
3301 self
._nsm
.nsr_update_cfg(msg
.id, msg
)
3304 self
._loop
.create_task(update_nsr_nsd(msg
.id, xact
, scratch
))
3306 for group
in msg
.scaling_group
:
3307 instance_delta
= get_scale_group_instance_delta(msg
.id, group
.scaling_group_name_ref
, xact
)
3308 self
._log
.debug("Got NSR:%s scale group instance delta: %s", msg
.id, instance_delta
)
3310 for instance_id
in instance_delta
["added"]:
3311 self
._nsm
.scale_nsr_out(msg
.id, group
.scaling_group_name_ref
, instance_id
, xact
)
3313 for instance_id
in instance_delta
["deleted"]:
3314 self
._nsm
.scale_nsr_in(msg
.id, group
.scaling_group_name_ref
, instance_id
)
3317 return RwTypes
.RwStatus
.SUCCESS
3320 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
3321 """ Prepare calllback from DTS for NSR """
3323 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3324 action
= xact_info
.query_action
3326 "Got Nsr prepare callback (xact: %s) (action: %s) (info: %s), %s:%s)",
3327 xact
, action
, xact_info
, xpath
, msg
3331 def delete_instantiation(ns_id
):
3332 """ Delete instantiation """
3333 yield from self
._nsm
.terminate_ns(ns_id
, None)
3335 def handle_delete_nsr():
3336 """ Handle delete NSR requests """
3337 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3338 # Terminate the NSR instance
3339 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3341 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3342 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3343 nsr
.record_event("terminate-rcvd", event_descr
)
3345 self
._loop
.create_task(delete_instantiation(msg
.id))
3347 fref
= ProtobufC
.FieldReference
.alloc()
3348 fref
.goto_whole_message(msg
.to_pbcm())
3350 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
, rwdts
.QueryAction
.DELETE
]:
3351 # if this is an NSR create
3352 if action
!= rwdts
.QueryAction
.DELETE
and msg
.id not in self
._nsm
.nsrs
:
3353 # Ensure the Cloud account/datacenter has been specified
3354 if not msg
.has_field("cloud_account") and not msg
.has_field("om_datacenter"):
3355 raise NsrInstantiationFailed("Cloud account or datacenter not specified in NSR")
3357 # Check if nsd is specified
3358 if not msg
.has_field("nsd"):
3359 raise NsrInstantiationFailed("NSD not specified in NSR")
3362 nsr
= self
._nsm
.nsrs
[msg
.id]
3364 if msg
.has_field("nsd"):
3365 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3366 raise NsrVlUpdateError("Unable to update VL when NSR not in running state")
3367 if 'vld' not in msg
.nsd
or len(msg
.nsd
.vld
) == 0:
3368 raise NsrVlUpdateError("NS config NSD should have atleast 1 VLD defined")
3370 if msg
.has_field("scaling_group"):
3371 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3372 raise ScalingOperationError("Unable to perform scaling action when NS is not in running state")
3374 if len(msg
.scaling_group
) > 1:
3375 raise ScalingOperationError("Only a single scaling group can be configured at a time")
3377 for group_msg
in msg
.scaling_group
:
3378 num_new_group_instances
= len(group_msg
.instance
)
3379 if num_new_group_instances
> 1:
3380 raise ScalingOperationError("Only a single scaling instance can be modified at a time")
3382 elif num_new_group_instances
== 1:
3383 scale_group
= nsr
.scaling_groups
[group_msg
.scaling_group_name_ref
]
3384 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
]:
3385 if len(scale_group
.instances
) == scale_group
.max_instance_count
:
3386 raise ScalingOperationError("Max instances for %s reached" % scale_group
)
3388 acg
.handle
.prepare_complete_ok(xact_info
.handle
)
3391 self
._log
.debug("Registering for NSR config using xpath: %s",
3392 NsrDtsHandler
.NSR_XPATH
)
3394 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
3395 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
3396 self
._nsr
_regh
= acg
.register(xpath
=NsrDtsHandler
.NSR_XPATH
,
3397 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3398 on_prepare
=on_prepare
)
3400 self
._scale
_regh
= acg
.register(
3401 xpath
=NsrDtsHandler
.SCALE_INSTANCE_XPATH
,
3402 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY| rwdts
.Flag
.CACHE
,
3405 self
._key
_pair
_regh
= acg
.register(
3406 xpath
=NsrDtsHandler
.KEY_PAIR_XPATH
,
3407 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3411 class NsrOpDataDtsHandler(object):
3412 """ The network service op data DTS handler """
3413 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
3415 def __init__(self
, dts
, log
, loop
, nsm
):
3424 """ Return the registration handle"""
3429 """ Return the NS manager instance """
3434 """ Register for Nsr op data publisher registration"""
3435 self
._log
.debug("Registering Nsr op data path %s as publisher",
3436 NsrOpDataDtsHandler
.XPATH
)
3438 hdl
= rift
.tasklets
.DTS
.RegistrationHandler()
3439 handlers
= rift
.tasklets
.Group
.Handler()
3440 with self
._dts
.group_create(handler
=handlers
) as group
:
3441 self
._regh
= group
.register(xpath
=NsrOpDataDtsHandler
.XPATH
,
3443 flags
=rwdts
.Flag
.PUBLISHER | rwdts
.Flag
.NO_PREP_READ | rwdts
.Flag
.DATASTORE
)
3446 def create(self
, path
, msg
):
3448 Create an NS record in DTS with the path and message
3450 self
._log
.debug("Creating NSR %s:%s", path
, msg
)
3451 self
.regh
.create_element(path
, msg
)
3452 self
._log
.debug("Created NSR, %s:%s", path
, msg
)
3455 def update(self
, path
, msg
, flags
=rwdts
.XactFlag
.REPLACE
):
3457 Update an NS record in DTS with the path and message
3459 self
._log
.debug("Updating NSR, %s:%s regh = %s", path
, msg
, self
.regh
)
3460 self
.regh
.update_element(path
, msg
, flags
)
3461 self
._log
.debug("Updated NSR, %s:%s", path
, msg
)
3464 def delete(self
, path
):
3466 Update an NS record in DTS with the path and message
3468 self
._log
.debug("Deleting NSR path:%s", path
)
3469 self
.regh
.delete_element(path
)
3470 self
._log
.debug("Deleted NSR path:%s", path
)
3473 class VnfrDtsHandler(object):
3474 """ The virtual network service DTS handler """
3475 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
3477 def __init__(self
, dts
, log
, loop
, nsm
):
3487 """ Return registration handle """
3492 """ Return the NS manager instance """
3497 """ Register for vnfr create/update/delete/ advises from dts """
3499 def on_commit(xact_info
):
3500 """ The transaction has been committed """
3501 self
._log
.debug("Got vnfr commit (xact_info: %s)", xact_info
)
3502 return rwdts
.MemberRspCode
.ACTION_OK
3505 def on_prepare(xact_info
, action
, ks_path
, msg
):
3506 """ prepare callback from dts """
3507 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3509 "Got vnfr on_prepare cb (xact_info: %s, action: %s): %s:%s",
3510 xact_info
, action
, ks_path
, msg
3513 schema
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr
.schema()
3514 path_entry
= schema
.keyspec_to_entry(ks_path
)
3515 if path_entry
.key00
.id not in self
._nsm
._vnfrs
:
3516 self
._log
.error("%s request for non existent record path %s",
3518 xact_info
.respond_xpath(rwdts
.XactRspCode
.NA
, xpath
)
3522 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3523 if action
== rwdts
.QueryAction
.CREATE
or action
== rwdts
.QueryAction
.UPDATE
:
3524 yield from self
._nsm
.update_vnfr(msg
)
3525 elif action
== rwdts
.QueryAction
.DELETE
:
3526 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3527 self
._nsm
.delete_vnfr(path_entry
.key00
.id)
3529 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
, xpath
)
3531 self
._log
.debug("Registering for VNFR using xpath: %s",
3532 VnfrDtsHandler
.XPATH
,)
3534 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_commit
=on_commit
,
3535 on_prepare
=on_prepare
,)
3536 with self
._dts
.group_create() as group
:
3537 self
._regh
= group
.register(xpath
=VnfrDtsHandler
.XPATH
,
3539 flags
=(rwdts
.Flag
.SUBSCRIBER
),)
3542 class NsdRefCountDtsHandler(object):
3543 """ The NSD Ref Count DTS handler """
3544 XPATH
= "D,/nsr:ns-instance-opdata/rw-nsr:nsd-ref-count"
3546 def __init__(self
, dts
, log
, loop
, nsm
):
3556 """ Return registration handle """
3561 """ Return the NS manager instance """
3566 """ Register for NSD ref count read from dts """
3569 def on_prepare(xact_info
, action
, ks_path
, msg
):
3570 """ prepare callback from dts """
3571 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3573 if action
== rwdts
.QueryAction
.READ
:
3574 schema
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount
.schema()
3575 path_entry
= schema
.keyspec_to_entry(ks_path
)
3576 nsd_list
= yield from self
._nsm
.get_nsd_refcount(path_entry
.key00
.nsd_id_ref
)
3577 for xpath
, msg
in nsd_list
:
3578 xact_info
.respond_xpath(rsp_code
=rwdts
.XactRspCode
.MORE
,
3581 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
3583 raise NetworkServiceRecordError("Not supported operation %s" % action
)
3585 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_prepare
,)
3586 with self
._dts
.group_create() as group
:
3587 self
._regh
= group
.register(xpath
=NsdRefCountDtsHandler
.XPATH
,
3589 flags
=rwdts
.Flag
.PUBLISHER
,)
3592 class NsManager(object):
3593 """ The Network Service Manager class"""
3594 def __init__(self
, dts
, log
, loop
,
3595 nsr_handler
, vnfr_handler
, vlr_handler
, ro_plugin_selector
,
3596 vnffgmgr
, vnfd_pub_handler
, cloud_account_handler
):
3600 self
._nsr
_handler
= nsr_handler
3601 self
._vnfr
_pub
_handler
= vnfr_handler
3602 self
._vlr
_pub
_handler
= vlr_handler
3603 self
._vnffgmgr
= vnffgmgr
3604 self
._vnfd
_pub
_handler
= vnfd_pub_handler
3605 self
._cloud
_account
_handler
= cloud_account_handler
3607 self
._ro
_plugin
_selector
= ro_plugin_selector
3608 self
._ncclient
= rift
.mano
.ncclient
.NcClient(
3620 self
.cfgmgr_obj
= conman
.ROConfigManager(log
, loop
, dts
, self
)
3622 # TODO: All these handlers should move to tasklet level.
3623 # Passing self is often an indication of bad design
3624 self
._nsd
_dts
_handler
= NsdDtsHandler(dts
, log
, loop
, self
)
3625 self
._vnfd
_dts
_handler
= VnfdDtsHandler(dts
, log
, loop
, self
)
3626 self
._dts
_handlers
= [self
._nsd
_dts
_handler
,
3627 VnfrDtsHandler(dts
, log
, loop
, self
),
3628 NsdRefCountDtsHandler(dts
, log
, loop
, self
),
3629 NsrDtsHandler(dts
, log
, loop
, self
),
3630 ScalingRpcHandler(log
, dts
, loop
, self
.scale_rpc_callback
),
3631 NsrRpcDtsHandler(dts
,log
,loop
,self
),
3632 self
._vnfd
_dts
_handler
,
3653 def nsr_handler(self
):
3654 """" NSR handler """
3655 return self
._nsr
_handler
3659 """" So Obj handler """
3664 """ NSRs in this NSM"""
3669 """ NSDs in this NSM"""
3674 """ VNFDs in this NSM"""
3679 """ VNFRs in this NSM"""
3683 def nsr_pub_handler(self
):
3684 """ NSR publication handler """
3685 return self
._nsr
_handler
3688 def vnfr_pub_handler(self
):
3689 """ VNFR publication handler """
3690 return self
._vnfr
_pub
_handler
3693 def vlr_pub_handler(self
):
3694 """ VLR publication handler """
3695 return self
._vlr
_pub
_handler
3698 def vnfd_pub_handler(self
):
3699 return self
._vnfd
_pub
_handler
3703 """ Register all static DTS handlers """
3704 for dts_handle
in self
._dts
_handlers
:
3705 yield from dts_handle
.register()
3708 def get_ns_by_nsr_id(self
, nsr_id
):
3709 """ get NSR by nsr id """
3710 if nsr_id
not in self
._nsrs
:
3711 raise NetworkServiceRecordError("NSR id %s not found" % nsr_id
)
3713 return self
._nsrs
[nsr_id
]
3715 def scale_nsr_out(self
, nsr_id
, scale_group_name
, instance_id
, config_xact
):
3716 self
.log
.debug("Scale out NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3721 nsr
= self
._nsrs
[nsr_id
]
3722 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3723 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3725 self
._loop
.create_task(nsr
.create_scale_group_instance(scale_group_name
, instance_id
, config_xact
))
3727 def scale_nsr_in(self
, nsr_id
, scale_group_name
, instance_id
):
3728 self
.log
.debug("Scale in NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3733 nsr
= self
._nsrs
[nsr_id
]
3734 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3735 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3737 self
._loop
.create_task(nsr
.delete_scale_group_instance(scale_group_name
, instance_id
))
3739 def scale_rpc_callback(self
, xact
, msg
, action
):
3740 """Callback handler for RPC calls
3742 xact : Transaction Handler
3744 action : Scaling Action
3746 ScalingGroupInstance
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup_Instance
3747 ScalingGroup
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
3749 xpath
= ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').format(
3751 instance
= ScalingGroupInstance
.from_dict({"id": msg
.instance_id
})
3754 def get_nsr_scaling_group():
3755 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3757 for result
in results
:
3758 res
= yield from result
3759 nsr_config
= res
.result
3761 for scaling_group
in nsr_config
.scaling_group
:
3762 if scaling_group
.scaling_group_name_ref
== msg
.scaling_group_name_ref
:
3765 scaling_group
= nsr_config
.scaling_group
.add()
3766 scaling_group
.scaling_group_name_ref
= msg
.scaling_group_name_ref
3768 return (nsr_config
, scaling_group
)
3771 def update_config(nsr_config
):
3772 xml
= self
._ncclient
.convert_to_xml(RwNsrYang
, nsr_config
)
3773 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
3774 yield from self
._ncclient
.connect()
3775 yield from self
._ncclient
.manager
.edit_config(target
="running", config
=xml
, default_operation
="replace")
3779 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3780 scaling_group
.instance
.append(instance
)
3781 yield from update_config(nsr_config
)
3785 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3786 scaling_group
.instance
.remove(instance
)
3787 yield from update_config(nsr_config
)
3789 if action
== ScalingRpcHandler
.ACTION
.SCALE_OUT
:
3790 self
._loop
.create_task(scale_out())
3792 self
._loop
.create_task(scale_in())
3794 # Opdata based calls, disabled for now!
3795 # if action == ScalingRpcHandler.ACTION.SCALE_OUT:
3796 # self.scale_nsr_out(
3798 # msg.scaling_group_name_ref,
3802 # self.scale_nsr_in(
3804 # msg.scaling_group_name_ref,
3807 def nsr_update_cfg(self
, nsr_id
, msg
):
3808 nsr
= self
._nsrs
[nsr_id
]
3809 nsr
.nsr_cfg_msg
= msg
3811 def nsr_instantiate_vl(self
, nsr_id
, vld
):
3812 self
.log
.debug("NSR {} create VL {}".format(nsr_id
, vld
))
3813 nsr
= self
._nsrs
[nsr_id
]
3814 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3815 raise NsrVlUpdateError("Cannot perform VL instantiate if NSR is not in running state")
3817 # Not calling in a separate task as this is called from a separate task
3818 yield from nsr
.create_vl_instance(vld
)
3820 def nsr_terminate_vl(self
, nsr_id
, vld
):
3821 self
.log
.debug("NSR {} delete VL {}".format(nsr_id
, vld
.id))
3822 nsr
= self
._nsrs
[nsr_id
]
3823 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3824 raise NsrVlUpdateError("Cannot perform VL terminate if NSR is not in running state")
3826 # Not calling in a separate task as this is called from a separate task
3827 yield from nsr
.delete_vl_instance(vld
)
3829 def create_nsr(self
, nsr_msg
, key_pairs
=None,restart_mode
=False):
3830 """ Create an NSR instance """
3831 if nsr_msg
.id in self
._nsrs
:
3832 msg
= "NSR id %s already exists" % nsr_msg
.id
3833 self
._log
.error(msg
)
3834 raise NetworkServiceRecordError(msg
)
3836 self
._log
.info("Create NetworkServiceRecord nsr id %s from nsd_id %s",
3840 nsm_plugin
= self
._ro
_plugin
_selector
.ro_plugin
3841 sdn_account_name
= self
._cloud
_account
_handler
.get_cloud_account_sdn_name(nsr_msg
.cloud_account
)
3843 nsr
= NetworkServiceRecord(self
._dts
,
3851 restart_mode
=restart_mode
,
3852 vlr_handler
=self
._ro
_plugin
_selector
._records
_publisher
._vlr
_pub
_hdlr
3854 self
._nsrs
[nsr_msg
.id] = nsr
3855 nsm_plugin
.create_nsr(nsr_msg
, nsr_msg
.nsd
, key_pairs
)
3859 def delete_nsr(self
, nsr_id
):
3861 Delete NSR with the passed nsr id
3863 del self
._nsrs
[nsr_id
]
3866 def instantiate_ns(self
, nsr_id
, config_xact
):
3867 """ Instantiate an NS instance """
3868 self
._log
.debug("Instantiating Network service id %s", nsr_id
)
3869 if nsr_id
not in self
._nsrs
:
3870 err
= "NSR id %s not found " % nsr_id
3871 self
._log
.error(err
)
3872 raise NetworkServiceRecordError(err
)
3874 nsr
= self
._nsrs
[nsr_id
]
3875 yield from nsr
.nsm_plugin
.instantiate_ns(nsr
, config_xact
)
3878 def update_vnfr(self
, vnfr
):
3879 """Create/Update an VNFR """
3881 vnfr_state
= self
._vnfrs
[vnfr
.id].state
3882 self
._log
.debug("Updating VNFR with state %s: vnfr %s", vnfr_state
, vnfr
)
3884 yield from self
._vnfrs
[vnfr
.id].update_state(vnfr
)
3885 nsr
= self
.find_nsr_for_vnfr(vnfr
.id)
3886 yield from nsr
.update_state()
3888 def find_nsr_for_vnfr(self
, vnfr_id
):
3889 """ Find the NSR which )has the passed vnfr id"""
3890 for nsr
in list(self
.nsrs
.values()):
3891 for vnfr
in list(nsr
.vnfrs
.values()):
3892 if vnfr
.id == vnfr_id
:
3896 def delete_vnfr(self
, vnfr_id
):
3897 """ Delete VNFR with the passed id"""
3898 del self
._vnfrs
[vnfr_id
]
3900 def get_nsd_ref(self
, nsd_id
):
3901 """ Get network service descriptor for the passed nsd_id
3903 nsd
= self
.get_nsd(nsd_id
)
3908 def get_nsr_config(self
, nsd_id
):
3909 xpath
= "C,/nsr:ns-instance-config"
3910 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3912 for result
in results
:
3913 entry
= yield from result
3914 ns_instance_config
= entry
.result
3916 for nsr
in ns_instance_config
.nsr
:
3917 if nsr
.nsd
.id == nsd_id
:
3923 def nsd_unref_by_nsr_id(self
, nsr_id
):
3924 """ Unref the network service descriptor based on NSR id """
3925 self
._log
.debug("NSR Unref called for Nsr Id:%s", nsr_id
)
3926 if nsr_id
in self
._nsrs
:
3927 nsr
= self
._nsrs
[nsr_id
]
3930 nsd
= self
.get_nsd(nsr
.nsd_id
)
3931 self
._log
.debug("Releasing ref on NSD %s held by NSR %s - Curr %d",
3932 nsd
.id, nsr
.id, nsd
.ref_count
)
3934 except NetworkServiceDescriptorError
:
3935 # We store a copy of NSD in NSR and the NSD in nsd-catalog
3940 self
._log
.error("Cannot find NSR with id %s", nsr_id
)
3941 raise NetworkServiceDescriptorUnrefError("No NSR with id" % nsr_id
)
3944 def nsd_unref(self
, nsd_id
):
3945 """ Unref the network service descriptor associated with the id """
3946 nsd
= self
.get_nsd(nsd_id
)
3949 def get_nsd(self
, nsd_id
):
3950 """ Get network service descriptor for the passed nsd_id"""
3951 if nsd_id
not in self
._nsds
:
3952 self
._log
.error("Cannot find NSD id:%s", nsd_id
)
3953 raise NetworkServiceDescriptorError("Cannot find NSD id:%s", nsd_id
)
3955 return self
._nsds
[nsd_id
]
3957 def create_nsd(self
, nsd_msg
):
3958 """ Create a network service descriptor """
3959 self
._log
.debug("Create network service descriptor - %s", nsd_msg
)
3960 if nsd_msg
.id in self
._nsds
:
3961 self
._log
.error("Cannot create NSD %s -NSD ID already exists", nsd_msg
)
3962 raise NetworkServiceDescriptorError("NSD already exists-%s", nsd_msg
.id)
3964 nsd
= NetworkServiceDescriptor(
3971 self
._nsds
[nsd_msg
.id] = nsd
3975 def update_nsd(self
, nsd
):
3976 """ update the Network service descriptor """
3977 self
._log
.debug("Update network service descriptor - %s", nsd
)
3978 if nsd
.id not in self
._nsds
:
3979 self
._log
.debug("No NSD found - creating NSD id = %s", nsd
.id)
3980 self
.create_nsd(nsd
)
3982 self
._log
.debug("Updating NSD id = %s, nsd = %s", nsd
.id, nsd
)
3983 self
._nsds
[nsd
.id].update(nsd
)
3985 def delete_nsd(self
, nsd_id
):
3986 """ Delete the Network service descriptor with the passed id """
3987 self
._log
.debug("Deleting the network service descriptor - %s", nsd_id
)
3988 if nsd_id
not in self
._nsds
:
3989 self
._log
.debug("Delete NSD failed - cannot find nsd-id %s", nsd_id
)
3990 raise NetworkServiceDescriptorNotFound("Cannot find %s", nsd_id
)
3992 if nsd_id
not in self
._nsds
:
3993 self
._log
.debug("Cannot delete NSD id %s reference exists %s",
3995 self
._nsds
[nsd_id
].ref_count
)
3996 raise NetworkServiceDescriptorRefCountExists(
3997 "Cannot delete :%s, ref_count:%s",
3999 self
._nsds
[nsd_id
].ref_count
)
4001 del self
._nsds
[nsd_id
]
4003 def get_vnfd_config(self
, xact
):
4004 vnfd_dts_reg
= self
._vnfd
_dts
_handler
.regh
4005 for cfg
in vnfd_dts_reg
.get_xact_elements(xact
):
4006 if cfg
.id not in self
._vnfds
:
4007 self
.create_vnfd(cfg
)
4009 def get_vnfd(self
, vnfd_id
, xact
):
4010 """ Get virtual network function descriptor for the passed vnfd_id"""
4011 if vnfd_id
not in self
._vnfds
:
4012 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4013 self
.get_vnfd_config(xact
)
4015 if vnfd_id
not in self
._vnfds
:
4016 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4017 raise VnfDescriptorError("Cannot find VNFD id:%s", vnfd_id
)
4019 return self
._vnfds
[vnfd_id
]
4021 def create_vnfd(self
, vnfd
):
4022 """ Create a virtual network function descriptor """
4023 self
._log
.debug("Create virtual network function descriptor - %s", vnfd
)
4024 if vnfd
.id in self
._vnfds
:
4025 self
._log
.error("Cannot create VNFD %s -VNFD ID already exists", vnfd
)
4026 raise VnfDescriptorError("VNFD already exists-%s", vnfd
.id)
4028 self
._vnfds
[vnfd
.id] = vnfd
4029 return self
._vnfds
[vnfd
.id]
4031 def update_vnfd(self
, vnfd
):
4032 """ Update the virtual network function descriptor """
4033 self
._log
.debug("Update virtual network function descriptor- %s", vnfd
)
4036 if vnfd
.id not in self
._vnfds
:
4037 self
._log
.debug("No VNFD found - creating VNFD id = %s", vnfd
.id)
4038 self
.create_vnfd(vnfd
)
4040 self
._log
.debug("Updating VNFD id = %s, vnfd = %s", vnfd
.id, vnfd
)
4041 self
._vnfds
[vnfd
.id] = vnfd
4044 def delete_vnfd(self
, vnfd_id
):
4045 """ Delete the virtual network function descriptor with the passed id """
4046 self
._log
.debug("Deleting the virtual network function descriptor - %s", vnfd_id
)
4047 if vnfd_id
not in self
._vnfds
:
4048 self
._log
.debug("Delete VNFD failed - cannot find vnfd-id %s", vnfd_id
)
4049 raise VnfDescriptorError("Cannot find %s", vnfd_id
)
4051 del self
._vnfds
[vnfd_id
]
4053 def nsd_in_use(self
, nsd_id
):
4054 """ Is the NSD with the passed id in use """
4055 self
._log
.debug("Is this NSD in use - msg:%s", nsd_id
)
4056 if nsd_id
in self
._nsds
:
4057 return self
._nsds
[nsd_id
].in_use()
4061 def publish_nsr(self
, xact
, path
, msg
):
4062 """ Publish a NSR """
4063 self
._log
.debug("Publish NSR with path %s, msg %s",
4065 yield from self
.nsr_handler
.update(xact
, path
, msg
)
4068 def unpublish_nsr(self
, xact
, path
):
4069 """ Un Publish an NSR """
4070 self
._log
.debug("Publishing delete NSR with path %s", path
)
4071 yield from self
.nsr_handler
.delete(path
, xact
)
4073 def vnfr_is_ready(self
, vnfr_id
):
4074 """ VNFR with the id is ready """
4075 self
._log
.debug("VNFR id %s ready", vnfr_id
)
4076 if vnfr_id
not in self
._vnfds
:
4077 err
= "Did not find VNFR ID with id %s" % vnfr_id
4078 self
._log
.critical("err")
4079 raise VirtualNetworkFunctionRecordError(err
)
4080 self
._vnfrs
[vnfr_id
].is_ready()
4083 def get_nsd_refcount(self
, nsd_id
):
4084 """ Get the nsd_list from this NSM"""
4086 def nsd_refcount_xpath(nsd_id
):
4087 """ xpath for ref count entry """
4088 return (NsdRefCountDtsHandler
.XPATH
+
4089 "[rw-nsr:nsd-id-ref = '{}']").format(nsd_id
)
4092 if nsd_id
is None or nsd_id
== "":
4093 for nsd
in self
._nsds
.values():
4094 nsd_msg
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4095 nsd_msg
.nsd_id_ref
= nsd
.id
4096 nsd_msg
.instance_ref_count
= nsd
.ref_count
4097 nsd_list
.append((nsd_refcount_xpath(nsd
.id), nsd_msg
))
4098 elif nsd_id
in self
._nsds
:
4099 nsd_msg
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4100 nsd_msg
.nsd_id_ref
= self
._nsds
[nsd_id
].id
4101 nsd_msg
.instance_ref_count
= self
._nsds
[nsd_id
].ref_count
4102 nsd_list
.append((nsd_refcount_xpath(nsd_id
), nsd_msg
))
4107 def terminate_ns(self
, nsr_id
, xact
):
4109 Terminate network service for the given NSR Id
4112 # Terminate the instances/networks assocaited with this nw service
4113 self
._log
.debug("Terminating the network service %s", nsr_id
)
4115 yield from self
._nsrs
[nsr_id
].terminate()
4116 except Exception as e
:
4117 self
.log
.exception("Failed to terminate NSR[id=%s]", nsr_id
)
4120 yield from self
.nsd_unref_by_nsr_id(nsr_id
)
4122 # Unpublish the NSR record
4123 self
._log
.debug("Unpublishing the network service %s", nsr_id
)
4124 yield from self
._nsrs
[nsr_id
].unpublish(xact
)
4126 # Finaly delete the NS instance from this NS Manager
4127 self
._log
.debug("Deletng the network service %s", nsr_id
)
4128 self
.delete_nsr(nsr_id
)
4131 class NsmRecordsPublisherProxy(object):
4132 """ This class provides a publisher interface that allows plugin objects
4133 to publish NSR/VNFR/VLR"""
4135 def __init__(self
, dts
, log
, loop
, nsr_pub_hdlr
, vnfr_pub_hdlr
, vlr_pub_hdlr
):
4139 self
._nsr
_pub
_hdlr
= nsr_pub_hdlr
4140 self
._vlr
_pub
_hdlr
= vlr_pub_hdlr
4141 self
._vnfr
_pub
_hdlr
= vnfr_pub_hdlr
4144 def publish_nsr(self
, xact
, nsr
):
4145 """ Publish an NSR """
4146 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4147 return (yield from self
._nsr
_pub
_hdlr
.update(xact
, path
, nsr
))
4150 def unpublish_nsr(self
, xact
, nsr
):
4151 """ Unpublish an NSR """
4152 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4153 return (yield from self
._nsr
_pub
_hdlr
.delete(xact
, path
))
4156 def publish_vnfr(self
, xact
, vnfr
):
4157 """ Publish an VNFR """
4158 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4159 return (yield from self
._vnfr
_pub
_hdlr
.update(xact
, path
, vnfr
))
4162 def unpublish_vnfr(self
, xact
, vnfr
):
4163 """ Unpublish a VNFR """
4164 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4165 return (yield from self
._vnfr
_pub
_hdlr
.delete(xact
, path
))
4168 def publish_vlr(self
, xact
, vlr
):
4169 """ Publish a VLR """
4170 path
= VirtualLinkRecord
.vlr_xpath(vlr
)
4171 return (yield from self
._vlr
_pub
_hdlr
.update(xact
, path
, vlr
))
4174 def unpublish_vlr(self
, xact
, vlr
):
4175 """ Unpublish a VLR """
4176 path
= VirtualLinkRecord
.vlr_xpath(vlr
)
4177 return (yield from self
._vlr
_pub
_hdlr
.delete(xact
, path
))
4180 class ScalingRpcHandler(mano_dts
.DtsHandler
):
4181 """ The Network service Monitor DTS handler """
4182 SCALE_IN_INPUT_XPATH
= "I,/nsr:exec-scale-in"
4183 SCALE_IN_OUTPUT_XPATH
= "O,/nsr:exec-scale-in"
4185 SCALE_OUT_INPUT_XPATH
= "I,/nsr:exec-scale-out"
4186 SCALE_OUT_OUTPUT_XPATH
= "O,/nsr:exec-scale-out"
4188 ACTION
= Enum('ACTION', 'SCALE_IN SCALE_OUT')
4190 def __init__(self
, log
, dts
, loop
, callback
=None):
4191 super().__init
__(log
, dts
, loop
)
4192 self
.callback
= callback
4193 self
.last_instance_id
= defaultdict(int)
4199 def on_scale_in_prepare(xact_info
, action
, ks_path
, msg
):
4200 assert action
== rwdts
.QueryAction
.RPC
4204 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_IN
)
4206 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleIn
.from_dict({
4207 "instance_id": msg
.instance_id
})
4209 xact_info
.respond_xpath(
4210 rwdts
.XactRspCode
.ACK
,
4211 self
.__class
__.SCALE_IN_OUTPUT_XPATH
,
4214 except Exception as e
:
4215 self
.log
.exception(e
)
4216 xact_info
.respond_xpath(
4217 rwdts
.XactRspCode
.NACK
,
4218 self
.__class
__.SCALE_IN_OUTPUT_XPATH
)
4221 def on_scale_out_prepare(xact_info
, action
, ks_path
, msg
):
4222 assert action
== rwdts
.QueryAction
.RPC
4225 scaling_group
= msg
.scaling_group_name_ref
4226 if not msg
.instance_id
:
4227 last_instance_id
= self
.last_instance_id
[scale_group
]
4228 msg
.instance_id
= last_instance_id
+ 1
4229 self
.last_instance_id
[scale_group
] += 1
4232 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_OUT
)
4234 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleOut
.from_dict({
4235 "instance_id": msg
.instance_id
})
4237 xact_info
.respond_xpath(
4238 rwdts
.XactRspCode
.ACK
,
4239 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
,
4242 except Exception as e
:
4243 self
.log
.exception(e
)
4244 xact_info
.respond_xpath(
4245 rwdts
.XactRspCode
.NACK
,
4246 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
)
4248 scale_in_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4249 on_prepare
=on_scale_in_prepare
)
4250 scale_out_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4251 on_prepare
=on_scale_out_prepare
)
4253 with self
.dts
.group_create() as group
:
4255 xpath
=self
.__class
__.SCALE_IN_INPUT_XPATH
,
4256 handler
=scale_in_hdl
,
4257 flags
=rwdts
.Flag
.PUBLISHER
)
4259 xpath
=self
.__class
__.SCALE_OUT_INPUT_XPATH
,
4260 handler
=scale_out_hdl
,
4261 flags
=rwdts
.Flag
.PUBLISHER
)
4264 class NsmTasklet(rift
.tasklets
.Tasklet
):
4266 The network service manager tasklet
4268 def __init__(self
, *args
, **kwargs
):
4269 super(NsmTasklet
, self
).__init
__(*args
, **kwargs
)
4270 self
.rwlog
.set_category("rw-mano-log")
4271 self
.rwlog
.set_subcategory("nsm")
4276 self
._ro
_plugin
_selector
= None
4277 self
._vnffgmgr
= None
4279 self
._nsr
_handler
= None
4280 self
._vnfr
_pub
_handler
= None
4281 self
._vlr
_pub
_handler
= None
4282 self
._vnfd
_pub
_handler
= None
4283 self
._scale
_cfg
_handler
= None
4285 self
._records
_publisher
_proxy
= None
4288 """ The task start callback """
4289 super(NsmTasklet
, self
).start()
4290 self
.log
.info("Starting NsmTasklet")
4292 self
.log
.debug("Registering with dts")
4293 self
._dts
= rift
.tasklets
.DTS(self
.tasklet_info
,
4294 RwNsmYang
.get_schema(),
4296 self
.on_dts_state_change
)
4298 self
.log
.debug("Created DTS Api GI Object: %s", self
._dts
)
4304 print("Caught Exception in NSM stop:", sys
.exc_info()[0])
4307 def on_instance_started(self
):
4308 """ Task instance started callback """
4309 self
.log
.debug("Got instance started callback")
4313 """ Task init callback """
4314 self
.log
.debug("Got instance started callback")
4316 self
.log
.debug("creating config account handler")
4318 self
._nsr
_pub
_handler
= publisher
.NsrOpDataDtsHandler(self
._dts
, self
.log
, self
.loop
)
4319 yield from self
._nsr
_pub
_handler
.register()
4321 self
._vnfr
_pub
_handler
= publisher
.VnfrPublisherDtsHandler(self
._dts
, self
.log
, self
.loop
)
4322 yield from self
._vnfr
_pub
_handler
.register()
4324 self
._vlr
_pub
_handler
= publisher
.VlrPublisherDtsHandler(self
._dts
, self
.log
, self
.loop
)
4325 yield from self
._vlr
_pub
_handler
.register()
4327 manifest
= self
.tasklet_info
.get_pb_manifest()
4328 use_ssl
= manifest
.bootstrap_phase
.rwsecurity
.use_ssl
4329 ssl_cert
= manifest
.bootstrap_phase
.rwsecurity
.cert
4330 ssl_key
= manifest
.bootstrap_phase
.rwsecurity
.key
4332 self
._vnfd
_pub
_handler
= publisher
.VnfdPublisher(use_ssl
, ssl_cert
, ssl_key
, self
.loop
)
4334 self
._records
_publisher
_proxy
= NsmRecordsPublisherProxy(
4338 self
._nsr
_pub
_handler
,
4339 self
._vnfr
_pub
_handler
,
4340 self
._vlr
_pub
_handler
,
4343 # Register the NSM to receive the nsm plugin
4344 # when cloud account is configured
4345 self
._ro
_plugin
_selector
= cloud
.ROAccountPluginSelector(
4349 self
._records
_publisher
_proxy
,
4351 yield from self
._ro
_plugin
_selector
.register()
4353 self
._cloud
_account
_handler
= cloud
.CloudAccountConfigSubscriber(
4358 yield from self
._cloud
_account
_handler
.register()
4360 self
._vnffgmgr
= rwvnffgmgr
.VnffgMgr(self
._dts
, self
.log
, self
.log_hdl
, self
.loop
)
4361 yield from self
._vnffgmgr
.register()
4363 self
._nsm
= NsManager(
4367 self
._nsr
_pub
_handler
,
4368 self
._vnfr
_pub
_handler
,
4369 self
._vlr
_pub
_handler
,
4370 self
._ro
_plugin
_selector
,
4372 self
._vnfd
_pub
_handler
,
4373 self
._cloud
_account
_handler
4376 yield from self
._nsm
.register()
4380 """ Task run callback """
4384 def on_dts_state_change(self
, state
):
4385 """Take action according to current dts state to transition
4386 application into the corresponding application state
4389 state - current dts state
4392 rwdts
.State
.INIT
: rwdts
.State
.REGN_COMPLETE
,
4393 rwdts
.State
.CONFIG
: rwdts
.State
.RUN
,
4397 rwdts
.State
.INIT
: self
.init
,
4398 rwdts
.State
.RUN
: self
.run
,
4401 # Transition application to next state
4402 handler
= handlers
.get(state
, None)
4403 if handler
is not None:
4404 yield from handler()
4406 # Transition dts to next state
4407 next_state
= switch
.get(state
, None)
4408 if next_state
is not None:
4409 self
.log
.debug("Changing state to %s", next_state
)
4410 self
._dts
.handle
.set_state(next_state
)