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 "nsd_id": self
._nsr
.nsd_id
,
224 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
225 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
226 "sdn_account": self
._sdn
_account
_name
,
227 "operational_status": 'init',
229 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
230 elif self
._vnffgr
_state
== VnffgRecordState
.TERMINATED
:
231 vnffgr_dict
= {"id": self
._vnffgr
_id
,
232 "nsd_id": self
._nsr
.nsd_id
,
233 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
234 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
235 "sdn_account": self
._sdn
_account
_name
,
236 "operational_status": 'terminated',
238 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
241 vnffgr
= self
._vnffgmgr
.fetch_vnffgr(self
._vnffgr
_id
)
243 self
._log
.exception("Fetching VNFFGR for VNFFG with id %s failed", self
._vnffgr
_id
)
244 self
._vnffgr
_state
= VnffgRecordState
.FAILED
245 vnffgr_dict
= {"id": self
._vnffgr
_id
,
246 "nsd_id": self
._nsr
.nsd_id
,
247 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
248 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
249 "sdn_account": self
._sdn
_account
_name
,
250 "operational_status": 'failed',
252 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
257 def vnffgr_create_msg(self
):
258 """ Virtual Link Record message for Creating VLR in VNS """
259 vnffgr_dict
= {"id": self
._vnffgr
_id
,
260 "nsd_id": self
._nsr
.nsd_id
,
261 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
262 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
263 "sdn_account": self
._sdn
_account
_name
,
265 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
266 for rsp
in self
._vnffgd
_msg
.rsp
:
267 vnffgr_rsp
= vnffgr
.rsp
.add()
268 vnffgr_rsp
.id = str(uuid
.uuid4())
269 vnffgr_rsp
.name
= self
._nsr
.name
+ '.' + rsp
.name
270 self
._vnffgr
_rsp
_id
.append(vnffgr_rsp
.id)
271 vnffgr_rsp
.vnffgd_rsp_id_ref
= rsp
.id
272 vnffgr_rsp
.vnffgd_rsp_name_ref
= rsp
.name
273 for rsp_cp_ref
in rsp
.vnfd_connection_point_ref
:
274 vnfd
= [vnfr
.vnfd
for vnfr
in self
._nsr
.vnfrs
.values() if vnfr
.vnfd
.id == rsp_cp_ref
.vnfd_id_ref
]
275 self
._log
.debug("VNFD message during VNFFG instantiation is %s",vnfd
)
276 if len(vnfd
) > 0 and vnfd
[0].has_field('service_function_type'):
277 self
._log
.debug("Service Function Type for VNFD ID %s is %s",rsp_cp_ref
.vnfd_id_ref
, vnfd
[0].service_function_type
)
279 self
._log
.error("Service Function Type not available for VNFD ID %s; Skipping in chain",rsp_cp_ref
.vnfd_id_ref
)
282 vnfr_cp_ref
= vnffgr_rsp
.vnfr_connection_point_ref
.add()
283 vnfr_cp_ref
.member_vnf_index_ref
= rsp_cp_ref
.member_vnf_index_ref
284 vnfr_cp_ref
.hop_number
= rsp_cp_ref
.order
285 vnfr_cp_ref
.vnfd_id_ref
=rsp_cp_ref
.vnfd_id_ref
286 vnfr_cp_ref
.service_function_type
= vnfd
[0].service_function_type
287 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
288 if (nsr_vnfr
.vnfd
.id == vnfr_cp_ref
.vnfd_id_ref
and
289 nsr_vnfr
.member_vnf_index
== vnfr_cp_ref
.member_vnf_index_ref
):
290 vnfr_cp_ref
.vnfr_id_ref
= nsr_vnfr
.id
291 vnfr_cp_ref
.vnfr_name_ref
= nsr_vnfr
.name
292 vnfr_cp_ref
.vnfr_connection_point_ref
= rsp_cp_ref
.vnfd_connection_point_ref
294 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
295 self
._log
.debug(" Received VNFR is %s", vnfr
)
296 while vnfr
.operational_status
!= 'running':
297 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
298 if vnfr
.operational_status
== 'failed':
299 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
300 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self
.id, vnfr
.id))
301 yield from asyncio
.sleep(2, loop
=self
._loop
)
302 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
303 self
._log
.debug("Received VNFR is %s", vnfr
)
305 vnfr_cp_ref
.connection_point_params
.mgmt_address
= vnfr
.mgmt_interface
.ip_address
306 for cp
in vnfr
.connection_point
:
307 if cp
.name
== vnfr_cp_ref
.vnfr_connection_point_ref
:
308 vnfr_cp_ref
.connection_point_params
.port_id
= cp
.connection_point_id
309 vnfr_cp_ref
.connection_point_params
.name
= self
._nsr
.name
+ '.' + cp
.name
310 for vdu
in vnfr
.vdur
:
311 for ext_intf
in vdu
.external_interface
:
312 if ext_intf
.name
== vnfr_cp_ref
.vnfr_connection_point_ref
:
313 vnfr_cp_ref
.connection_point_params
.vm_id
= vdu
.vim_id
314 self
._log
.debug("VIM ID for CP %s in VNFR %s is %s",cp
.name
,nsr_vnfr
.id,
315 vnfr_cp_ref
.connection_point_params
.vm_id
)
318 vnfr_cp_ref
.connection_point_params
.address
= cp
.ip_address
319 vnfr_cp_ref
.connection_point_params
.port
= VnffgRecord
.SFF_DP_PORT
321 for vnffgd_classifier
in self
._vnffgd
_msg
.classifier
:
322 _rsp
= [rsp
for rsp
in vnffgr
.rsp
if rsp
.vnffgd_rsp_id_ref
== vnffgd_classifier
.rsp_id_ref
]
324 rsp_id_ref
= _rsp
[0].id
325 rsp_name
= _rsp
[0].name
327 self
._log
.error("RSP with ID %s not found during classifier creation for classifier id %s",vnffgd_classifier
.rsp_id_ref
,vnffgd_classifier
.id)
329 vnffgr_classifier
= vnffgr
.classifier
.add()
330 vnffgr_classifier
.id = vnffgd_classifier
.id
331 vnffgr_classifier
.name
= self
._nsr
.name
+ '.' + vnffgd_classifier
.name
332 _rsp
[0].classifier_name
= vnffgr_classifier
.name
333 vnffgr_classifier
.rsp_id_ref
= rsp_id_ref
334 vnffgr_classifier
.rsp_name
= rsp_name
335 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
336 if (nsr_vnfr
.vnfd
.id == vnffgd_classifier
.vnfd_id_ref
and
337 nsr_vnfr
.member_vnf_index
== vnffgd_classifier
.member_vnf_index_ref
):
338 vnffgr_classifier
.vnfr_id_ref
= nsr_vnfr
.id
339 vnffgr_classifier
.vnfr_name_ref
= nsr_vnfr
.name
340 vnffgr_classifier
.vnfr_connection_point_ref
= vnffgd_classifier
.vnfd_connection_point_ref
342 if nsr_vnfr
.vnfd
.service_function_chain
== 'CLASSIFIER':
343 vnffgr_classifier
.sff_name
= nsr_vnfr
.name
345 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
346 self
._log
.debug(" Received VNFR is %s", vnfr
)
347 while vnfr
.operational_status
!= 'running':
348 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
349 if vnfr
.operational_status
== 'failed':
350 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
351 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self
.id, vnfr
.id))
352 yield from asyncio
.sleep(2, loop
=self
._loop
)
353 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
354 self
._log
.debug("Received VNFR is %s", vnfr
)
356 for cp
in vnfr
.connection_point
:
357 if cp
.name
== vnffgr_classifier
.vnfr_connection_point_ref
:
358 vnffgr_classifier
.port_id
= cp
.connection_point_id
359 vnffgr_classifier
.ip_address
= cp
.ip_address
360 for vdu
in vnfr
.vdur
:
361 for ext_intf
in vdu
.external_interface
:
362 if ext_intf
.name
== vnffgr_classifier
.vnfr_connection_point_ref
:
363 vnffgr_classifier
.vm_id
= vdu
.vim_id
364 self
._log
.debug("VIM ID for CP %s in VNFR %s is %s",cp
.name
,nsr_vnfr
.id,
365 vnfr_cp_ref
.connection_point_params
.vm_id
)
368 self
._log
.info("VNFFGR msg to be sent is %s", vnffgr
)
372 def vnffgr_nsr_sff_list(self
):
373 """ SFF List for VNFR """
375 sf_list
= [nsr_vnfr
.name
for nsr_vnfr
in self
._nsr
.vnfrs
.values() if nsr_vnfr
.vnfd
.service_function_chain
== 'SF']
377 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
378 if (nsr_vnfr
.vnfd
.service_function_chain
== 'CLASSIFIER' or nsr_vnfr
.vnfd
.service_function_chain
== 'SFF'):
379 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
380 self
._log
.debug(" Received VNFR is %s", vnfr
)
381 while vnfr
.operational_status
!= 'running':
382 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
383 if vnfr
.operational_status
== 'failed':
384 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
385 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self
.id, vnfr
.id))
386 yield from asyncio
.sleep(2, loop
=self
._loop
)
387 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
388 self
._log
.debug("Received VNFR is %s", vnfr
)
390 sff
= RwsdnYang
.VNFFGSff()
391 sff_list
[nsr_vnfr
.vnfd
.id] = sff
392 sff
.name
= nsr_vnfr
.name
393 sff
.function_type
= nsr_vnfr
.vnfd
.service_function_chain
395 sff
.mgmt_address
= vnfr
.mgmt_interface
.ip_address
396 sff
.mgmt_port
= VnffgRecord
.SFF_MGMT_PORT
397 for cp
in vnfr
.connection_point
:
398 sff_dp
= sff
.dp_endpoints
.add()
399 sff_dp
.name
= self
._nsr
.name
+ '.' + cp
.name
400 sff_dp
.address
= cp
.ip_address
401 sff_dp
.port
= VnffgRecord
.SFF_DP_PORT
402 if nsr_vnfr
.vnfd
.service_function_chain
== 'SFF':
403 for sf_name
in sf_list
:
404 _sf
= sff
.vnfr_list
.add()
405 _sf
.vnfr_name
= sf_name
410 def instantiate(self
):
411 """ Instantiate this VNFFG """
413 self
._log
.info("Instaniating VNFFGR with vnffgd %s",
417 vnffgr_request
= yield from self
.vnffgr_create_msg()
418 vnffg_sff_list
= yield from self
.vnffgr_nsr_sff_list()
421 vnffgr
= self
._vnffgmgr
.create_vnffgr(vnffgr_request
,self
._vnffgd
_msg
.classifier
,vnffg_sff_list
)
422 except Exception as e
:
423 self
._log
.exception("VNFFG instantiation failed: %s", str(e
))
424 self
._vnffgr
_state
= VnffgRecordState
.FAILED
425 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFFGR %s failure" % (self
.id, vnffgr_request
.id))
427 self
._vnffgr
_state
= VnffgRecordState
.INSTANTIATION_PENDING
429 self
._log
.info("Instantiated VNFFGR :%s", vnffgr
)
430 self
._vnffgr
_state
= VnffgRecordState
.ACTIVE
432 self
._log
.info("Invoking update_state to update NSR state for NSR ID: %s", self
._nsr
.id)
433 yield from self
._nsr
.update_state()
435 def vnffgr_in_vnffgrm(self
):
436 """ Is there a VNFR record in VNFM """
437 if (self
._vnffgr
_state
== VnffgRecordState
.ACTIVE
or
438 self
._vnffgr
_state
== VnffgRecordState
.INSTANTIATION_PENDING
or
439 self
._vnffgr
_state
== VnffgRecordState
.FAILED
):
446 """ Terminate this VNFFGR """
447 if not self
.vnffgr_in_vnffgrm():
448 self
._log
.error("Ignoring terminate request for id %s in state %s",
449 self
.id, self
._vnffgr
_state
)
452 self
._log
.info("Terminating VNFFGR id:%s", self
.id)
453 self
._vnffgr
_state
= VnffgRecordState
.TERMINATE_PENDING
455 self
._vnffgmgr
.terminate_vnffgr(self
._vnffgr
_id
)
457 self
._vnffgr
_state
= VnffgRecordState
.TERMINATED
458 self
._log
.debug("Terminated VNFFGR id:%s", self
.id)
461 class VirtualLinkRecord(object):
462 """ Virtual Link Records class"""
463 XPATH
= "D,/vlr:vlr-catalog/vlr:vlr"
466 def create_record(dts
, log
, loop
, nsr_name
, vld_msg
, cloud_account_name
, om_datacenter
, ip_profile
, nsr_id
, restart_mode
=False):
467 """Creates a new VLR object based on the given data.
469 If restart mode is enabled, then we look for existing records in the
470 DTS and create a VLR records using the exiting data(ID)
475 vlr_obj
= VirtualLinkRecord(
488 res_iter
= yield from dts
.query_read(
489 "D,/vlr:vlr-catalog/vlr:vlr",
490 rwdts
.XactFlag
.MERGE
)
493 response
= yield from fut
494 vlr
= response
.result
496 # Check if the record is already present, if so use the ID of
497 # the existing record. Since the name of the record is uniquely
498 # formed we can use it as a search key!
499 if vlr
.name
== vlr_obj
.name
:
500 vlr_obj
.reset_id(vlr
.id)
505 def __init__(self
, dts
, log
, loop
, nsr_name
, vld_msg
, cloud_account_name
, om_datacenter
, ip_profile
, nsr_id
):
509 self
._nsr
_name
= nsr_name
510 self
._vld
_msg
= vld_msg
511 self
._cloud
_account
_name
= cloud_account_name
512 self
._om
_datacenter
_name
= om_datacenter
513 self
._assigned
_subnet
= None
514 self
._nsr
_id
= nsr_id
515 self
._ip
_profile
= ip_profile
516 self
._vlr
_id
= str(uuid
.uuid4())
517 self
._state
= VlRecordState
.INIT
518 self
._prev
_state
= None
519 self
._create
_time
= int(time
.time())
523 """ path for this object """
524 return "D,/vlr:vlr-catalog/vlr:vlr[vlr:id = '{}']".format(self
._vlr
_id
)
533 """ Get NSR name for this VL """
538 """ Virtual Link Desciptor """
542 def assigned_subnet(self
):
543 """ Subnet assigned to this VL"""
544 return self
._assigned
_subnet
549 Get the name for this VLR.
550 VLR name is "nsr name:VLD name"
552 if self
.vld_msg
.vim_network_name
:
553 return self
.vld_msg
.vim_network_name
554 elif self
.vld_msg
.name
== "multisite":
555 # This is a temporary hack to identify manually provisioned inter-site network
556 return self
.vld_msg
.name
558 return self
._nsr
_name
+ "." + self
.vld_msg
.name
561 def cloud_account_name(self
):
562 """ Cloud account that this VLR should be created in """
563 return self
._cloud
_account
_name
566 def om_datacenter_name(self
):
567 """ Datacenter that this VLR should be created in """
568 return self
._om
_datacenter
_name
572 """ Get the VLR path from VLR """
573 return (VirtualLinkRecord
.XPATH
+ "[vlr:id = '{}']").format(vlr
.id)
581 def state(self
, value
):
582 """ VLR set state """
586 def prev_state(self
):
587 """ VLR previous state """
588 return self
._prev
_state
591 def prev_state(self
, value
):
592 """ VLR set previous state """
593 self
._prev
_state
= value
597 """ Virtual Link Record message for Creating VLR in VNS """
598 vld_fields
= ["short_name",
606 vld_copy_dict
= {k
: v
for k
, v
in self
.vld_msg
.as_dict().items()
609 vlr_dict
= {"id": self
._vlr
_id
,
610 "nsr_id_ref": self
._nsr
_id
,
611 "vld_ref": self
.vld_msg
.id,
613 "create_time": self
._create
_time
,
614 "cloud_account": self
.cloud_account_name
,
615 "om_datacenter": self
.om_datacenter_name
,
618 if self
._ip
_profile
and self
._ip
_profile
.has_field('ip_profile_params'):
619 vlr_dict
['ip_profile_params' ] = self
._ip
_profile
.ip_profile_params
.as_dict()
621 vlr_dict
.update(vld_copy_dict
)
622 vlr
= RwVlrYang
.YangData_Vlr_VlrCatalog_Vlr
.from_dict(vlr_dict
)
625 def reset_id(self
, vlr_id
):
626 self
._vlr
_id
= vlr_id
628 def create_nsr_vlr_msg(self
, vnfrs
):
629 """ The VLR message"""
630 nsr_vlr
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vlr()
631 nsr_vlr
.vlr_ref
= self
._vlr
_id
632 nsr_vlr
.assigned_subnet
= self
.assigned_subnet
633 nsr_vlr
.cloud_account
= self
.cloud_account_name
634 nsr_vlr
.om_datacenter
= self
.om_datacenter_name
636 for conn
in self
.vld_msg
.vnfd_connection_point_ref
:
638 if (vnfr
.vnfd
.id == conn
.vnfd_id_ref
and
639 vnfr
.member_vnf_index
== conn
.member_vnf_index_ref
and
640 self
.cloud_account_name
== vnfr
.cloud_account_name
and
641 self
.om_datacenter_name
== vnfr
.om_datacenter_name
):
642 cp_entry
= nsr_vlr
.vnfr_connection_point_ref
.add()
643 cp_entry
.vnfr_id
= vnfr
.id
644 cp_entry
.connection_point
= conn
.vnfd_connection_point_ref
649 def instantiate(self
):
650 """ Instantiate this VL """
651 self
._log
.debug("Instaniating VLR key %s, vld %s",
652 self
.xpath
, self
._vld
_msg
)
654 self
._state
= VlRecordState
.INSTANTIATION_PENDING
655 self
._log
.debug("Executing VL create path:%s msg:%s",
656 self
.xpath
, self
.vlr_msg
)
658 with self
._dts
.transaction(flags
=0) as xact
:
659 block
= xact
.block_create()
660 block
.add_query_create(self
.xpath
, self
.vlr_msg
)
661 self
._log
.debug("Executing VL create path:%s msg:%s",
662 self
.xpath
, self
.vlr_msg
)
663 res_iter
= yield from block
.execute(now
=True)
669 self
._state
= VlRecordState
.FAILED
670 raise NsrInstantiationFailed("Failed NS %s instantiation due to empty response" % self
.id)
672 if vlr
.operational_status
== 'failed':
673 self
._log
.debug("NS Id:%s VL creation failed for vlr id %s", self
.id, vlr
.id)
674 self
._state
= VlRecordState
.FAILED
675 raise NsrInstantiationFailed("Failed VL %s instantiation (%s)" % (vlr
.id, vlr
.operational_status_details
))
677 self
._log
.info("Instantiated VL with xpath %s and vlr:%s",
679 self
._state
= VlRecordState
.ACTIVE
680 self
._assigned
_subnet
= vlr
.assigned_subnet
682 def vlr_in_vns(self
):
683 """ Is there a VLR record in VNS """
684 if (self
._state
== VlRecordState
.ACTIVE
or
685 self
._state
== VlRecordState
.INSTANTIATION_PENDING
or
686 self
._state
== VlRecordState
.TERMINATE_PENDING
or
687 self
._state
== VlRecordState
.FAILED
):
694 """ Terminate this VL """
695 if not self
.vlr_in_vns():
696 self
._log
.debug("Ignoring terminate request for id %s in state %s",
697 self
.id, self
._state
)
700 self
._log
.debug("Terminating VL id:%s", self
.id)
701 self
._state
= VlRecordState
.TERMINATE_PENDING
703 with self
._dts
.transaction(flags
=0) as xact
:
704 block
= xact
.block_create()
705 block
.add_query_delete(self
.xpath
)
706 yield from block
.execute(flags
=0, now
=True)
708 self
._state
= VlRecordState
.TERMINATED
709 self
._log
.debug("Terminated VL id:%s", self
.id)
712 class VnfRecordState(Enum
):
713 """ Vnf Record State """
715 INSTANTIATION_PENDING
= 102
717 TERMINATE_PENDING
= 104
722 class VirtualNetworkFunctionRecord(object):
723 """ Virtual Network Function Record class"""
724 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
728 def create_record(dts
, log
, loop
, vnfd
, const_vnfd_msg
, nsd_id
, nsr_name
,
729 cloud_account_name
, om_datacenter_name
, nsr_id
, group_name
, group_instance_id
,
730 placement_groups
, restart_mode
=False):
731 """Creates a new VNFR object based on the given data.
733 If restart mode is enabled, then we look for existing records in the
734 DTS and create a VNFR records using the exiting data(ID)
737 VirtualNetworkFunctionRecord
739 vnfr_obj
= VirtualNetworkFunctionRecord(
753 restart_mode
=restart_mode
)
756 res_iter
= yield from dts
.query_read(
757 "D,/vnfr:vnfr-catalog/vnfr:vnfr",
758 rwdts
.XactFlag
.MERGE
)
761 response
= yield from fut
762 vnfr
= response
.result
764 if vnfr
.name
== vnfr_obj
.name
:
765 vnfr_obj
.reset_id(vnfr
.id)
782 group_instance_id
=None,
783 placement_groups
= [],
784 restart_mode
= False):
789 self
._const
_vnfd
_msg
= const_vnfd_msg
790 self
._nsd
_id
= nsd_id
791 self
._nsr
_name
= nsr_name
792 self
._nsr
_id
= nsr_id
793 self
._cloud
_account
_name
= cloud_account_name
794 self
._om
_datacenter
_name
= om_datacenter_name
795 self
._group
_name
= group_name
796 self
._group
_instance
_id
= group_instance_id
797 self
._placement
_groups
= placement_groups
798 self
._config
_status
= NsrYang
.ConfigStates
.INIT
799 self
._create
_time
= int(time
.time())
801 self
._prev
_state
= VnfRecordState
.INIT
802 self
._state
= VnfRecordState
.INIT
803 self
._state
_failed
_reason
= None
805 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
808 self
._vnfr
_id
= str(uuid
.uuid4())
810 self
._vnfr
_msg
= self
.create_vnfr_msg()
811 self
._log
.debug("Set VNFR {} config type to {}".
812 format(self
.name
, self
.config_type
))
813 self
.restart_mode
= restart_mode
816 if group_name
is None and group_instance_id
is not None:
817 raise ValueError("Group instance id must not be provided with an empty group name")
827 return "D,/vnfr:vnfr-catalog/vnfr:vnfr[vnfr:id = '{}']".format(self
.id)
832 return self
._vnfr
_msg
835 def const_vnfr_msg(self
):
837 return RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_ConstituentVnfrRef(vnfr_id
=self
.id,cloud_account
=self
.cloud_account_name
,om_datacenter
=self
._om
_datacenter
_name
)
845 def cloud_account_name(self
):
846 """ Cloud account that this VNF should be created in """
847 return self
._cloud
_account
_name
850 def om_datacenter_name(self
):
851 """ Datacenter that this VNF should be created in """
852 return self
._om
_datacenter
_name
857 """ Is this VNF actve """
858 return True if self
._state
== VnfRecordState
.ACTIVE
else False
862 """ state of this VNF """
866 def state_failed_reason(self
):
867 """ Error message in case this VNF is in failed state """
868 return self
._state
_failed
_reason
871 def member_vnf_index(self
):
872 """ Member VNF index """
873 return self
._const
_vnfd
_msg
.member_vnf_index
878 return self
._nsr
_name
882 """ Name of this VNFR """
883 if self
._name
is not None:
886 name_tags
= [self
._nsr
_name
]
888 if self
._group
_name
is not None:
889 name_tags
.append(self
._group
_name
)
891 if self
._group
_instance
_id
is not None:
892 name_tags
.append(str(self
._group
_instance
_id
))
894 name_tags
.extend([self
.vnfd
.name
, str(self
.member_vnf_index
)])
896 self
._name
= "__".join(name_tags
)
901 def vnfr_xpath(vnfr
):
902 """ Get the VNFR path from VNFR """
903 return (VirtualNetworkFunctionRecord
.XPATH
+ "[vnfr:id = '{}']").format(vnfr
.id)
906 def config_type(self
):
907 cfg_types
= ['netconf', 'juju', 'script']
908 for method
in cfg_types
:
909 if self
._vnfd
.vnf_configuration
.has_field(method
):
914 def config_status(self
):
915 """Return the config status as YANG ENUM string"""
916 self
._log
.debug("Map VNFR {} config status {} ({})".
917 format(self
.name
, self
._config
_status
, self
.config_type
))
918 if self
.config_type
== 'none':
919 return 'config_not_needed'
920 elif self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
922 elif self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
927 def set_state(self
, state
):
928 """ set the state of this object """
929 self
._prev
_state
= self
._state
932 def reset_id(self
, vnfr_id
):
933 self
._vnfr
_id
= vnfr_id
934 self
._vnfr
_msg
= self
.create_vnfr_msg()
937 self
.config_store
.merge_vnfd_config(
940 self
.member_vnf_index
,
943 def create_vnfr_msg(self
):
944 """ VNFR message for this VNFR """
952 vnfd_copy_dict
= {k
: v
for k
, v
in self
._vnfd
.as_dict().items() if k
in vnfd_fields
}
955 "nsr_id_ref": self
._nsr
_id
,
957 "cloud_account": self
._cloud
_account
_name
,
958 "om_datacenter": self
._om
_datacenter
_name
,
959 "config_status": self
.config_status
961 vnfr_dict
.update(vnfd_copy_dict
)
963 vnfr
= RwVnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr
.from_dict(vnfr_dict
)
965 vnfr
.vnfd
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_Vnfd
.from_dict(self
.vnfd
.as_dict())
966 vnfr
.member_vnf_index_ref
= self
.member_vnf_index
967 vnfr
.vnf_configuration
.from_dict(self
._vnfd
.vnf_configuration
.as_dict())
969 if self
._vnfd
.mgmt_interface
.has_field("port"):
970 vnfr
.mgmt_interface
.port
= self
._vnfd
.mgmt_interface
.port
972 for group_info
in self
._placement
_groups
:
973 group
= vnfr
.placement_groups_info
.add()
974 group
.from_dict(group_info
.as_dict())
976 # UI expects the monitoring param field to exist
977 vnfr
.monitoring_param
= []
979 self
._log
.debug("Get vnfr_msg for VNFR {} : {}".format(self
.name
, vnfr
))
983 def update_vnfm(self
):
984 self
._log
.debug("Send an update to VNFM for VNFR {} with {}".
985 format(self
.name
, self
.vnfr_msg
))
986 yield from self
._dts
.query_update(
988 rwdts
.XactFlag
.TRACE
,
992 def get_config_status(self
):
993 """Return the config status as YANG ENUM"""
994 return self
._config
_status
997 def set_config_status(self
, status
):
999 def status_to_string(status
):
1001 NsrYang
.ConfigStates
.INIT
: 'init',
1002 NsrYang
.ConfigStates
.CONFIGURING
: 'configuring',
1003 NsrYang
.ConfigStates
.CONFIG_NOT_NEEDED
: 'config_not_needed',
1004 NsrYang
.ConfigStates
.CONFIGURED
: 'configured',
1005 NsrYang
.ConfigStates
.FAILED
: 'failed',
1008 return status_dc
[status
]
1010 self
._log
.debug("Update VNFR {} from {} ({}) to {}".
1011 format(self
.name
, self
._config
_status
,
1012 self
.config_type
, status
))
1013 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1014 self
._log
.error("Updating already configured VNFR {}".
1018 if self
._config
_status
!= status
:
1020 self
._config
_status
= status
1021 # I don't think this is used. Original implementor can check.
1022 # Caused Exception, so corrected it by status_to_string
1023 # But not sure whats the use of this variable?
1024 self
.vnfr_msg
.config_status
= status_to_string(status
)
1025 except Exception as e
:
1026 self
._log
.error("Exception=%s", str(e
))
1029 self
._log
.debug("Updated VNFR {} status to {}".format(self
.name
, status
))
1031 if self
._config
_status
!= NsrYang
.ConfigStates
.INIT
:
1033 # Publish only after VNFM has the VNFR created
1034 yield from self
.update_vnfm()
1035 except Exception as e
:
1036 self
._log
.error("Exception updating VNFM with new status {} of VNFR {}: {}".
1037 format(status
, self
.name
, e
))
1038 self
._log
.exception(e
)
1040 def is_configured(self
):
1041 if self
.config_type
== 'none':
1044 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1050 def instantiate(self
, nsr
):
1051 """ Instantiate this VNFR"""
1053 self
._log
.debug("Instaniating VNFR key %s, vnfd %s",
1054 self
.xpath
, self
._vnfd
)
1056 self
._log
.debug("Create VNF with xpath %s and vnfr %s",
1057 self
.xpath
, self
.vnfr_msg
)
1059 self
.set_state(VnfRecordState
.INSTANTIATION_PENDING
)
1061 def find_vlr_for_cp(conn
):
1062 """ Find VLR for the given connection point """
1063 for vlr
in nsr
.vlrs
:
1064 for vnfd_cp
in vlr
.vld_msg
.vnfd_connection_point_ref
:
1065 if (vnfd_cp
.vnfd_id_ref
== self
._vnfd
.id and
1066 vnfd_cp
.vnfd_connection_point_ref
== conn
.name
and
1067 vnfd_cp
.member_vnf_index_ref
== self
.member_vnf_index
and
1068 vlr
.cloud_account_name
== self
.cloud_account_name
):
1069 self
._log
.debug("Found VLR for cp_name:%s and vnf-index:%d",
1070 conn
.name
, self
.member_vnf_index
)
1074 # For every connection point in the VNFD fill in the identifier
1075 for conn_p
in self
._vnfd
.connection_point
:
1076 cpr
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_ConnectionPoint()
1077 cpr
.name
= conn_p
.name
1078 cpr
.type_yang
= conn_p
.type_yang
1079 vlr_ref
= find_vlr_for_cp(conn_p
)
1081 msg
= "Failed to find VLR for cp = %s" % conn_p
.name
1082 self
._log
.debug("%s", msg
)
1083 # raise VirtualNetworkFunctionRecordError(msg)
1086 cpr
.vlr_ref
= vlr_ref
.id
1087 self
.vnfr_msg
.connection_point
.append(cpr
)
1088 self
._log
.debug("Connection point [%s] added, vnf id=%s vnfd id=%s",
1089 cpr
, self
.vnfr_msg
.id, self
.vnfr_msg
.vnfd
.id)
1091 if not self
.restart_mode
:
1092 yield from self
._dts
.query_create(self
.xpath
,
1096 yield from self
._dts
.query_update(self
.xpath
,
1100 self
._log
.info("Created VNF with xpath %s and vnfr %s",
1101 self
.xpath
, self
.vnfr_msg
)
1103 self
._log
.info("Instantiated VNFR with xpath %s and vnfd %s, vnfr %s",
1104 self
.xpath
, self
._vnfd
, self
.vnfr_msg
)
1107 def update_state(self
, vnfr_msg
):
1108 """ Update this VNFR"""
1109 if vnfr_msg
.operational_status
== "running":
1110 if self
.vnfr_msg
.operational_status
!= "running":
1111 yield from self
.is_active()
1112 elif vnfr_msg
.operational_status
== "failed":
1113 yield from self
.instantiation_failed(failed_reason
=vnfr_msg
.operational_status_details
)
1116 def is_active(self
):
1117 """ This VNFR is active """
1118 self
._log
.debug("VNFR %s is active", self
._vnfr
_id
)
1119 self
.set_state(VnfRecordState
.ACTIVE
)
1122 def instantiation_failed(self
, failed_reason
=None):
1123 """ This VNFR instantiation failed"""
1124 self
._log
.error("VNFR %s instantiation failed", self
._vnfr
_id
)
1125 self
.set_state(VnfRecordState
.FAILED
)
1126 self
._state
_failed
_reason
= failed_reason
1128 def vnfr_in_vnfm(self
):
1129 """ Is there a VNFR record in VNFM """
1130 if (self
._state
== VnfRecordState
.ACTIVE
or
1131 self
._state
== VnfRecordState
.INSTANTIATION_PENDING
or
1132 self
._state
== VnfRecordState
.FAILED
):
1138 def terminate(self
):
1139 """ Terminate this VNF """
1140 if not self
.vnfr_in_vnfm():
1141 self
._log
.debug("Ignoring terminate request for id %s in state %s",
1142 self
.id, self
._state
)
1145 self
._log
.debug("Terminating VNF id:%s", self
.id)
1146 self
.set_state(VnfRecordState
.TERMINATE_PENDING
)
1147 with self
._dts
.transaction(flags
=0) as xact
:
1148 block
= xact
.block_create()
1149 block
.add_query_delete(self
.xpath
)
1150 yield from block
.execute(flags
=0)
1151 self
.set_state(VnfRecordState
.TERMINATED
)
1152 self
._log
.debug("Terminated VNF id:%s", self
.id)
1155 class NetworkServiceStatus(object):
1156 """ A class representing the Network service's status """
1157 MAX_EVENTS_RECORDED
= 10
1158 """ Network service Status class"""
1159 def __init__(self
, dts
, log
, loop
):
1164 self
._state
= NetworkServiceRecordState
.INIT
1165 self
._events
= deque([])
1168 def create_notification(self
, evt
, evt_desc
, evt_details
):
1169 xp
= "N,/rw-nsr:nsm-notification"
1170 notif
= RwNsrYang
.YangNotif_RwNsr_NsmNotification()
1172 notif
.description
= evt_desc
1173 notif
.details
= evt_details
if evt_details
is not None else None
1175 yield from self
._dts
.query_create(xp
, rwdts
.XactFlag
.ADVISE
, notif
)
1176 self
._log
.info("Notification called by creating dts query: %s", notif
)
1178 def record_event(self
, evt
, evt_desc
, evt_details
):
1179 """ Record an event """
1180 self
._log
.debug("Recording event - evt %s, evt_descr %s len = %s",
1181 evt
, evt_desc
, len(self
._events
))
1182 if len(self
._events
) >= NetworkServiceStatus
.MAX_EVENTS_RECORDED
:
1183 self
._events
.popleft()
1184 self
._events
.append((int(time
.time()), evt
, evt_desc
,
1185 evt_details
if evt_details
is not None else None))
1187 self
._loop
.create_task(self
.create_notification(evt
,evt_desc
,evt_details
))
1189 def set_state(self
, state
):
1190 """ set the state of this status object """
1194 """ Return the state as a yang enum string """
1195 state_to_str_map
= {"INIT": "init",
1196 "VL_INIT_PHASE": "vl_init_phase",
1197 "VNF_INIT_PHASE": "vnf_init_phase",
1198 "VNFFG_INIT_PHASE": "vnffg_init_phase",
1199 "SCALING_GROUP_INIT_PHASE": "scaling_group_init_phase",
1200 "RUNNING": "running",
1201 "SCALING_OUT": "scaling_out",
1202 "SCALING_IN": "scaling_in",
1203 "TERMINATE_RCVD": "terminate_rcvd",
1204 "TERMINATE": "terminate",
1205 "VL_TERMINATE_PHASE": "vl_terminate_phase",
1206 "VNF_TERMINATE_PHASE": "vnf_terminate_phase",
1207 "VNFFG_TERMINATE_PHASE": "vnffg_terminate_phase",
1208 "TERMINATED": "terminated",
1210 "VL_INSTANTIATE": "vl_instantiate",
1211 "VL_TERMINATE": "vl_terminate",
1213 return state_to_str_map
[self
._state
.name
]
1217 """ State of this status object """
1222 """ Network Service Record as a message"""
1225 for entry
in self
._events
:
1226 event
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_OperationalEvents()
1229 event
.timestamp
, event
.event
, event
.description
, event
.details
= entry
1230 event_list
.append(event
)
1234 class NetworkServiceRecord(object):
1235 """ Network service record """
1236 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
1238 def __init__(self
, dts
, log
, loop
, nsm
, nsm_plugin
, nsr_cfg_msg
, sdn_account_name
, key_pairs
, restart_mode
=False,
1244 self
._nsr
_cfg
_msg
= nsr_cfg_msg
1245 self
._nsm
_plugin
= nsm_plugin
1246 self
._sdn
_account
_name
= sdn_account_name
1247 self
._vlr
_handler
= vlr_handler
1250 self
._nsr
_msg
= None
1251 self
._nsr
_regh
= None
1252 self
._key
_pairs
= key_pairs
1257 self
._param
_pools
= {}
1258 self
._scaling
_groups
= {}
1259 self
._create
_time
= int(time
.time())
1260 self
._op
_status
= NetworkServiceStatus(dts
, log
, loop
)
1261 self
._config
_status
= NsrYang
.ConfigStates
.CONFIGURING
1262 self
._config
_status
_details
= None
1264 self
.restart_mode
= restart_mode
1265 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
1266 self
._debug
_running
= False
1267 self
._is
_active
= False
1268 self
._vl
_phase
_completed
= False
1269 self
._vnf
_phase
_completed
= False
1270 self
.vlr_uptime_tasks
= {}
1273 # Initalise the state to init
1274 # The NSR moves through the following transitions
1275 # 1. INIT -> VLS_READY once all the VLs in the NSD are created
1276 # 2. VLS_READY - VNFS_READY when all the VNFs in the NSD are created
1277 # 3. VNFS_READY - READY when the NSR is published
1279 self
.set_state(NetworkServiceRecordState
.INIT
)
1281 self
.substitute_input_parameters
= InputParameterSubstitution(self
._log
)
1284 def nsm_plugin(self
):
1286 return self
._nsm
_plugin
1288 def set_state(self
, state
):
1289 """ Set state for this NSR"""
1290 self
._log
.debug("Setting state to %s", state
)
1291 # We are in init phase and is moving to the next state
1292 # The new state could be a FAILED state or VNF_INIIT_PHASE
1293 if self
.state
== NetworkServiceRecordState
.VL_INIT_PHASE
:
1294 self
._vl
_phase
_completed
= True
1296 if self
.state
== NetworkServiceRecordState
.VNF_INIT_PHASE
:
1297 self
._vnf
_phase
_completed
= True
1299 self
._op
_status
.set_state(state
)
1303 """ Get id for this NSR"""
1304 return self
._nsr
_cfg
_msg
.id
1308 """ Name of this network service record """
1309 return self
._nsr
_cfg
_msg
.name
1312 def cloud_account_name(self
):
1313 return self
._nsr
_cfg
_msg
.cloud_account
1316 def om_datacenter_name(self
):
1317 if self
._nsr
_cfg
_msg
.has_field('om_datacenter'):
1318 return self
._nsr
_cfg
_msg
.om_datacenter
1323 """State of this NetworkServiceRecord"""
1324 return self
._op
_status
.state
1328 """ Is this NSR active ?"""
1329 return True if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
else False
1333 """ VLRs associated with this NSR"""
1338 """ VNFRs associated with this NSR"""
1343 """ VNFFGRs associated with this NSR"""
1344 return self
._vnffgrs
1347 def scaling_groups(self
):
1348 """ Scaling groups associated with this NSR """
1349 return self
._scaling
_groups
1352 def param_pools(self
):
1353 """ Parameter value pools associated with this NSR"""
1354 return self
._param
_pools
1357 def nsr_cfg_msg(self
):
1358 return self
._nsr
_cfg
_msg
1361 def nsr_cfg_msg(self
, msg
):
1362 self
._nsr
_cfg
_msg
= msg
1366 """ NSD Protobuf for this NSR """
1367 if self
._nsd
is not None:
1369 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
1374 """ NSD ID for this NSR """
1375 return self
.nsd_msg
.id
1379 ''' Get a new job id for config primitive'''
1384 def config_status(self
):
1385 """ Config status for NSR """
1386 return self
._config
_status
1388 def resolve_placement_group_cloud_construct(self
, input_group
):
1390 Returns the cloud specific construct for placement group
1392 copy_dict
= ['name', 'requirement', 'strategy']
1394 for group_info
in self
._nsr
_cfg
_msg
.nsd_placement_group_maps
:
1395 if group_info
.placement_group_ref
== input_group
.name
:
1396 group
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_PlacementGroupsInfo()
1397 group_dict
= {k
:v
for k
,v
in
1398 group_info
.as_dict().items() if k
!= 'placement_group_ref'}
1399 for param
in copy_dict
:
1400 group_dict
.update({param
: getattr(input_group
, param
)})
1401 group
.from_dict(group_dict
)
1407 return "NSR(name={}, nsd_id={}, cloud_account={})".format(
1408 self
.name
, self
.nsd_id
, self
.cloud_account_name
1411 def _get_vnfd(self
, vnfd_id
, config_xact
):
1412 """ Fetch vnfd msg for the passed vnfd id """
1413 return self
._nsm
.get_vnfd(vnfd_id
, config_xact
)
1415 def _get_vnfd_cloud_account(self
, vnfd_member_index
):
1416 """ Fetch Cloud Account for the passed vnfd id """
1417 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1418 vim_accounts
= [(vnf
.cloud_account
,vnf
.om_datacenter
) for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map \
1419 if vnfd_member_index
== vnf
.member_vnf_index_ref
]
1420 if vim_accounts
and vim_accounts
[0]:
1421 return vim_accounts
[0]
1422 return (self
.cloud_account_name
,self
.om_datacenter_name
)
1424 def _get_constituent_vnfd_msg(self
, vnf_index
):
1425 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1426 if const_vnfd
.member_vnf_index
== vnf_index
:
1429 raise ValueError("Constituent VNF index %s not found" % vnf_index
)
1431 def record_event(self
, evt
, evt_desc
, evt_details
=None, state
=None):
1432 """ Record an event """
1433 self
._op
_status
.record_event(evt
, evt_desc
, evt_details
)
1434 if state
is not None:
1435 self
.set_state(state
)
1437 def scaling_trigger_str(self
, trigger
):
1438 SCALING_TRIGGER_STRS
= {
1439 NsdYang
.ScalingTrigger
.PRE_SCALE_IN
: 'pre-scale-in',
1440 NsdYang
.ScalingTrigger
.POST_SCALE_IN
: 'post-scale-in',
1441 NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
: 'pre-scale-out',
1442 NsdYang
.ScalingTrigger
.POST_SCALE_OUT
: 'post-scale-out',
1445 return SCALING_TRIGGER_STRS
[trigger
]
1446 except Exception as e
:
1447 self
._log
.error("Scaling trigger mapping error for {} : {}".
1449 self
._log
.exception(e
)
1450 return "Unknown trigger"
1453 def instantiate_vls(self
):
1455 This function instantiates VLs for every VL in this Network Service
1457 self
._log
.debug("Instantiating %d VLs in NSD id %s", len(self
._vlrs
),
1459 for vlr
in self
._vlrs
:
1460 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1461 vlr
.state
= VlRecordState
.ACTIVE
1462 self
.vlr_uptime_tasks
[vlr
.id] = self
._loop
.create_task(self
.vlr_uptime_update(vlr
))
1465 def vlr_uptime_update(self
, vlr
):
1468 vlr_
= RwVlrYang
.YangData_Vlr_VlrCatalog_Vlr
.from_dict({'id': vlr
.id})
1470 vlr_
.uptime
= int(time
.time()) - vlr
._create
_time
1471 yield from self
._vlr
_handler
.update(None, VirtualLinkRecord
.vlr_xpath(vlr
), vlr_
)
1472 yield from asyncio
.sleep(2, loop
=self
._loop
)
1473 except asyncio
.CancelledError
:
1474 self
._log
.debug("Received cancellation request for vlr_uptime_update task")
1475 yield from self
._vlr
_handler
.delete(None, VirtualLinkRecord
.vlr_xpath(vlr
))
1479 def create(self
, config_xact
):
1480 """ Create this network service"""
1481 # Create virtual links for all the external vnf
1482 # connection points in this NS
1483 yield from self
.create_vls()
1485 # Create VNFs in this network service
1486 yield from self
.create_vnfs(config_xact
)
1488 # Create VNFFG for network service
1489 self
.create_vnffgs()
1491 # Create Scaling Groups for each scaling group in NSD
1492 self
.create_scaling_groups()
1494 # Create Parameter Pools
1495 self
.create_param_pools()
1498 def apply_scale_group_config_script(self
, script
, group
, scale_instance
, trigger
, vnfrs
=None):
1499 """ Apply config based on script for scale group """
1502 def add_vnfrs_data(vnfrs_list
):
1503 """ Add as a dict each of the VNFRs data """
1505 for vnfr
in vnfrs_list
:
1506 self
._log
.debug("Add VNFR {} data".format(vnfr
))
1508 vnfr_data
['name'] = vnfr
.name
1509 if trigger
in [NsdYang
.ScalingTrigger
.PRE_SCALE_IN
, NsdYang
.ScalingTrigger
.POST_SCALE_OUT
]:
1510 # Get VNF management and other IPs, etc
1511 opdata
= yield from self
.fetch_vnfr(vnfr
.xpath
)
1512 self
._log
.debug("VNFR {} op data: {}".format(vnfr
.name
, opdata
))
1514 vnfr_data
['rw_mgmt_ip'] = opdata
.mgmt_interface
.ip_address
1515 vnfr_data
['rw_mgmt_port'] = opdata
.mgmt_interface
.port
1516 except Exception as e
:
1517 self
._log
.error("Unable to get management IP for vnfr {}:{}".
1518 format(vnfr
.name
, e
))
1521 vnfr_data
['connection_points'] = []
1522 for cp
in opdata
.connection_point
:
1524 con_pt
['name'] = cp
.name
1525 con_pt
['ip_address'] = cp
.ip_address
1526 vnfr_data
['connection_points'].append(con_pt
)
1527 except Exception as e
:
1528 self
._log
.error("Exception getting connections points for VNFR {}: {}".
1529 format(vnfr
.name
, e
))
1531 vnfrs_data
.append(vnfr_data
)
1532 self
._log
.debug("VNFRs data: {}".format(vnfrs_data
))
1536 def add_nsr_data(nsr
):
1538 nsr_data
['name'] = nsr
.name
1541 if script
is None or len(script
) == 0:
1542 self
._log
.error("Script not provided for scale group config: {}".format(group
.name
))
1545 if script
[0] == '/':
1548 path
= os
.path
.join(os
.environ
['RIFT_INSTALL'], "usr/bin", script
)
1549 if not os
.path
.exists(path
):
1550 self
._log
.error("Config faled for scale group {}: Script does not exist at {}".
1551 format(group
.name
, path
))
1554 # Build a YAML file with all parameters for the script to execute
1555 # The data consists of 5 sections
1557 # 2. Scale group config
1558 # 3. VNFRs in the scale group
1559 # 4. VNFRs outside scale group
1562 data
['trigger'] = group
.trigger_map(trigger
)
1563 data
['config'] = group
.group_msg
.as_dict()
1566 data
["vnfrs_in_group"] = yield from add_vnfrs_data(vnfrs
)
1568 data
["vnfrs_in_group"] = yield from add_vnfrs_data(scale_instance
.vnfrs
)
1570 data
["vnfrs_others"] = yield from add_vnfrs_data(self
.vnfrs
.values())
1571 data
["nsr"] = add_nsr_data(self
)
1574 with tempfile
.NamedTemporaryFile(delete
=False) as tmp_file
:
1575 tmp_file
.write(yaml
.dump(data
, default_flow_style
=True)
1578 self
._log
.debug("Creating a temp file: {} with input data: {}".
1579 format(tmp_file
.name
, data
))
1581 cmd
= "{} {}".format(path
, tmp_file
.name
)
1582 self
._log
.debug("Running the CMD: {}".format(cmd
))
1583 proc
= yield from asyncio
.create_subprocess_shell(cmd
, loop
=self
._loop
)
1584 rc
= yield from proc
.wait()
1586 self
._log
.error("The script {} for scale group {} config returned: {}".
1587 format(script
, group
.name
, rc
))
1595 def apply_scaling_group_config(self
, trigger
, group
, scale_instance
, vnfrs
=None):
1596 """ Apply the config for the scaling group based on trigger """
1597 if group
is None or scale_instance
is None:
1601 def update_config_status(success
=True, err_msg
=None):
1602 self
._log
.debug("Update %s config status to %r : %s",
1603 scale_instance
, success
, err_msg
)
1604 if (scale_instance
.config_status
== "failed"):
1605 # Do not update the config status if it is already in failed state
1608 if scale_instance
.config_status
== "configured":
1609 # Update only to failed state an already configured scale instance
1611 scale_instance
.config_status
= "failed"
1612 scale_instance
.config_err_msg
= err_msg
1613 yield from self
.update_state()
1615 # We are in configuring state
1616 # Only after post scale out mark instance as configured
1617 if trigger
== NsdYang
.ScalingTrigger
.POST_SCALE_OUT
:
1619 scale_instance
.config_status
= "configured"
1621 scale_instance
.config_status
= "failed"
1622 scale_instance
.config_err_msg
= err_msg
1623 yield from self
.update_state()
1625 config
= group
.trigger_config(trigger
)
1629 self
._log
.debug("Scaling group {} config: {}".format(group
.name
, config
))
1630 if config
.has_field("ns_config_primitive_name_ref"):
1631 config_name
= config
.ns_config_primitive_name_ref
1632 nsd_msg
= self
.nsd_msg
1633 config_primitive
= None
1634 for ns_cfg_prim
in nsd_msg
.service_primitive
:
1635 if ns_cfg_prim
.name
== config_name
:
1636 config_primitive
= ns_cfg_prim
1639 if config_primitive
is None:
1640 raise ValueError("Could not find ns_cfg_prim %s in nsr %s" % (config_name
, self
.name
))
1642 self
._log
.debug("Scaling group {} config primitive: {}".format(group
.name
, config_primitive
))
1643 if config_primitive
.has_field("user_defined_script"):
1644 rc
= yield from self
.apply_scale_group_config_script(config_primitive
.user_defined_script
,
1645 group
, scale_instance
, trigger
, vnfrs
)
1648 err_msg
= "Failed config for trigger {} using config script '{}'". \
1649 format(self
.scaling_trigger_str(trigger
),
1650 config_primitive
.user_defined_script
)
1651 yield from update_config_status(success
=rc
, err_msg
=err_msg
)
1654 err_msg
= "Failed config for trigger {} as config script is not specified". \
1655 format(self
.scaling_trigger_str(trigger
))
1656 yield from update_config_status(success
=False, err_msg
=err_msg
)
1657 raise NotImplementedError("Only script based config support for scale group for now: {}".
1660 err_msg
= "Failed config for trigger {} as config primitive is not specified".\
1661 format(self
.scaling_trigger_str(trigger
))
1662 yield from update_config_status(success
=False, err_msg
=err_msg
)
1663 self
._log
.error("Config primitive not specified for config action in scale group %s" %
1667 def create_scaling_groups(self
):
1668 """ This function creates a NSScalingGroup for every scaling
1669 group defined in he NSD"""
1671 for scaling_group_msg
in self
.nsd_msg
.scaling_group_descriptor
:
1672 self
._log
.debug("Found scaling_group %s in nsr id %s",
1673 scaling_group_msg
.name
, self
.id)
1675 group_record
= scale_group
.ScalingGroup(
1680 self
._scaling
_groups
[group_record
.name
] = group_record
1683 def create_scale_group_instance(self
, group_name
, index
, config_xact
, is_default
=False):
1684 group
= self
._scaling
_groups
[group_name
]
1685 scale_instance
= group
.create_instance(index
, is_default
)
1689 self
._log
.debug("Creating %u VNFs associated with NS id %s scaling group %s",
1690 len(self
.nsd_msg
.constituent_vnfd
), self
.id, self
)
1693 for vnf_index
, count
in group
.vnf_index_count_map
.items():
1694 const_vnfd_msg
= self
._get
_constituent
_vnfd
_msg
(vnf_index
)
1695 vnfd_msg
= self
._get
_vnfd
(const_vnfd_msg
.vnfd_id_ref
, config_xact
)
1697 cloud_account_name
, om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd_msg
.member_vnf_index
)
1698 if cloud_account_name
is None:
1699 cloud_account_name
= self
.cloud_account_name
1700 for _
in range(count
):
1701 vnfr
= yield from self
.create_vnf_record(vnfd_msg
, const_vnfd_msg
, cloud_account_name
, om_datacenter_name
, group_name
, index
)
1702 scale_instance
.add_vnfr(vnfr
)
1708 def instantiate_instance():
1709 self
._log
.debug("Creating %s VNFRS", scale_instance
)
1710 vnfrs
= yield from create_vnfs()
1711 yield from self
.publish()
1713 self
._log
.debug("Instantiating %s VNFRS for %s", len(vnfrs
), scale_instance
)
1714 scale_instance
.operational_status
= "vnf_init_phase"
1715 yield from self
.update_state()
1718 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
,
1719 group
, scale_instance
, vnfrs
)
1721 self
._log
.error("Pre scale out config for scale group {} ({}) failed".
1722 format(group
.name
, index
))
1723 scale_instance
.operational_status
= "failed"
1725 yield from self
.instantiate_vnfs(vnfrs
)
1727 except Exception as e
:
1728 self
._log
.exception("Failed to begin instantiatiation of vnfs for scale group {}: {}".
1729 format(group
.name
, e
))
1730 self
._log
.exception(e
)
1731 scale_instance
.operational_status
= "failed"
1733 yield from self
.update_state()
1735 yield from instantiate_instance()
1738 def delete_scale_group_instance(self
, group_name
, index
):
1739 group
= self
._scaling
_groups
[group_name
]
1740 scale_instance
= group
.get_instance(index
)
1741 if scale_instance
.is_default
:
1742 raise ScalingOperationError("Cannot terminate a default scaling group instance")
1744 scale_instance
.operational_status
= "terminate"
1745 yield from self
.update_state()
1748 def terminate_instance():
1749 self
._log
.debug("Terminating %s VNFRS" % scale_instance
)
1750 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_IN
,
1751 group
, scale_instance
)
1753 self
._log
.error("Pre scale in config for scale group {} ({}) failed".
1754 format(group
.name
, index
))
1756 # Going ahead with terminate, even if there is an error in pre-scale-in config
1757 # as this could be result of scale out failure and we need to cleanup this group
1758 yield from self
.terminate_vnfrs(scale_instance
.vnfrs
)
1759 group
.delete_instance(index
)
1761 scale_instance
.operational_status
= "vnf_terminate_phase"
1762 yield from self
.update_state()
1764 yield from terminate_instance()
1767 def _update_scale_group_instances_status(self
):
1769 def post_scale_out_task(group
, instance
):
1770 # Apply post scale out config once all VNFRs are active
1771 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_OUT
,
1773 instance
.operational_status
= "running"
1775 self
._log
.debug("Scale out for group {} and instance {} succeeded".
1776 format(group
.name
, instance
.instance_id
))
1778 self
._log
.error("Post scale out config for scale group {} ({}) failed".
1779 format(group
.name
, instance
.instance_id
))
1781 yield from self
.update_state()
1783 group_instances
= {group
: group
.instances
for group
in self
._scaling
_groups
.values()}
1784 for group
, instances
in group_instances
.items():
1785 self
._log
.debug("Updating %s instance status", group
)
1786 for instance
in instances
:
1787 instance_vnf_state_list
= [vnfr
.state
for vnfr
in instance
.vnfrs
]
1788 self
._log
.debug("Got vnfr instance states: %s", instance_vnf_state_list
)
1789 if instance
.operational_status
== "vnf_init_phase":
1790 if all([state
== VnfRecordState
.ACTIVE
for state
in instance_vnf_state_list
]):
1791 instance
.operational_status
= "running"
1793 # Create a task for post scale out to allow us to sleep before attempting
1794 # to configure newly created VM's
1795 self
._loop
.create_task(post_scale_out_task(group
, instance
))
1797 elif any([state
== VnfRecordState
.FAILED
for state
in instance_vnf_state_list
]):
1798 self
._log
.debug("Scale out for group {} and instance {} failed".
1799 format(group
.name
, instance
.instance_id
))
1800 instance
.operational_status
= "failed"
1802 elif instance
.operational_status
== "vnf_terminate_phase":
1803 if all([state
== VnfRecordState
.TERMINATED
for state
in instance_vnf_state_list
]):
1804 instance
.operational_status
= "terminated"
1805 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_IN
,
1808 self
._log
.debug("Scale in for group {} and instance {} succeeded".
1809 format(group
.name
, instance
.instance_id
))
1811 self
._log
.error("Post scale in config for scale group {} ({}) failed".
1812 format(group
.name
, instance
.instance_id
))
1814 def create_vnffgs(self
):
1815 """ This function creates VNFFGs for every VNFFG in the NSD
1816 associated with this NSR"""
1818 for vnffgd
in self
.nsd_msg
.vnffgd
:
1819 self
._log
.debug("Found vnffgd %s in nsr id %s", vnffgd
, self
.id)
1820 vnffgr
= VnffgRecord(self
._dts
,
1823 self
._nsm
._vnffgmgr
,
1827 self
._sdn
_account
_name
1829 self
._vnffgrs
[vnffgr
.id] = vnffgr
1831 def resolve_vld_ip_profile(self
, nsd_msg
, vld
):
1832 self
._log
.debug("Receieved ip profile ref is %s",vld
.ip_profile_ref
)
1833 if not vld
.has_field('ip_profile_ref'):
1835 profile
= [profile
for profile
in nsd_msg
.ip_profiles
if profile
.name
== vld
.ip_profile_ref
]
1836 return profile
[0] if profile
else None
1839 def _create_vls(self
, vld
, cloud_account
,om_datacenter
):
1840 """Create a VLR in the cloud account specified using the given VLD
1844 cloud_account : Cloud account name
1849 vlr
= yield from VirtualLinkRecord
.create_record(
1857 self
.resolve_vld_ip_profile(self
.nsd_msg
, vld
),
1859 restart_mode
=self
.restart_mode
)
1863 def _extract_cloud_accounts_for_vl(self
, vld
):
1865 Extracts the list of cloud accounts from the NS Config obj
1868 1. Cloud accounts based connection point (vnf_cloud_account_map)
1870 vld : VLD yang object
1875 cloud_account_list
= []
1877 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1878 # Handle case where cloud_account is None
1880 for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1881 if vnf
.cloud_account
is not None or vnf
.om_datacenter
is not None:
1882 vnf_cloud_map
[vnf
.member_vnf_index_ref
] = (vnf
.cloud_account
,vnf
.om_datacenter
)
1884 for vnfc
in vld
.vnfd_connection_point_ref
:
1885 cloud_account
= vnf_cloud_map
.get(
1886 vnfc
.member_vnf_index_ref
,
1887 (self
.cloud_account_name
,self
.om_datacenter_name
))
1889 cloud_account_list
.append(cloud_account
)
1891 if self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1892 for vld_map
in self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1893 if vld_map
.vld_id_ref
== vld
.id:
1894 for cloud_account
in vld_map
.cloud_accounts
:
1895 cloud_account_list
.extend((cloud_account
,None))
1896 for om_datacenter
in vld_map
.om_datacenters
:
1897 cloud_account_list
.extend((None,om_datacenter
))
1899 # If no config has been provided then fall-back to the default
1901 if not cloud_account_list
:
1902 cloud_account_list
= [(self
.cloud_account_name
,self
.om_datacenter_name
)]
1904 self
._log
.debug("VL {} cloud accounts: {}".
1905 format(vld
.name
, cloud_account_list
))
1906 return set(cloud_account_list
)
1909 def create_vls(self
):
1910 """ This function creates VLs for every VLD in the NSD
1911 associated with this NSR"""
1912 for vld
in self
.nsd_msg
.vld
:
1914 self
._log
.debug("Found vld %s in nsr id %s", vld
, self
.id)
1915 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1916 for cloud_account
,om_datacenter
in cloud_account_list
:
1917 vlr
= yield from self
._create
_vls
(vld
, cloud_account
,om_datacenter
)
1918 self
._vlrs
.append(vlr
)
1922 def create_vl_instance(self
, vld
):
1923 self
._log
.debug("Create VL for {}: {}".format(self
.id, vld
.as_dict()))
1924 # Check if the VL is already present
1926 for vl
in self
._vlrs
:
1927 if vl
.vld_msg
.id == vld
.id:
1928 self
._log
.debug("The VLD %s already in NSR %s as VLR %s with status %s",
1929 vld
.id, self
.id, vl
.id, vl
.state
)
1931 if vlr
.state
!= VlRecordState
.TERMINATED
:
1932 err_msg
= "VLR for VL %s in NSR %s already instantiated", \
1934 self
._log
.error(err_msg
)
1935 raise NsrVlUpdateError(err_msg
)
1939 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1940 for account
,om_datacenter
in cloud_account_list
:
1941 vlr
= yield from self
._create
_vls
(vld
, account
,om_datacenter
)
1942 self
._vlrs
.append(vlr
)
1944 vlr
.state
= VlRecordState
.INSTANTIATION_PENDING
1945 yield from self
.update_state()
1948 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1949 vlr
.state
= VlRecordState
.ACTIVE
1951 except Exception as e
:
1952 err_msg
= "Error instantiating VL for NSR {} and VLD {}: {}". \
1953 format(self
.id, vld
.id, e
)
1954 self
._log
.error(err_msg
)
1955 self
._log
.exception(e
)
1956 vlr
.state
= VlRecordState
.FAILED
1958 yield from self
.update_state()
1961 def delete_vl_instance(self
, vld
):
1962 for vlr
in self
._vlrs
:
1963 if vlr
.vld_msg
.id == vld
.id:
1964 self
._log
.debug("Found VLR %s for VLD %s in NSR %s",
1965 vlr
.id, vld
.id, self
.id)
1966 vlr
.state
= VlRecordState
.TERMINATE_PENDING
1967 yield from self
.update_state()
1970 yield from self
.nsm_plugin
.terminate_vl(vlr
)
1971 vlr
.state
= VlRecordState
.TERMINATED
1972 self
._vlrs
.remove(vlr
)
1974 except Exception as e
:
1975 err_msg
= "Error terminating VL for NSR {} and VLD {}: {}". \
1976 format(self
.id, vld
.id, e
)
1977 self
._log
.error(err_msg
)
1978 self
._log
.exception(e
)
1979 vlr
.state
= VlRecordState
.FAILED
1981 yield from self
.update_state()
1985 def create_vnfs(self
, config_xact
):
1987 This function creates VNFs for every VNF in the NSD
1988 associated with this NSR
1990 self
._log
.debug("Creating %u VNFs associated with this NS id %s",
1991 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
1993 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1994 if not const_vnfd
.start_by_default
:
1995 self
._log
.debug("start_by_default set to False in constituent VNF (%s). Skipping start.",
1996 const_vnfd
.member_vnf_index
)
1999 vnfd_msg
= self
._get
_vnfd
(const_vnfd
.vnfd_id_ref
, config_xact
)
2000 cloud_account_name
,om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd
.member_vnf_index
)
2001 if cloud_account_name
is None:
2002 cloud_account_name
= self
.cloud_account_name
2003 yield from self
.create_vnf_record(vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
)
2006 def get_placement_groups(self
, vnfd_msg
, const_vnfd
):
2007 placement_groups
= []
2008 for group
in self
.nsd_msg
.placement_groups
:
2009 for member_vnfd
in group
.member_vnfd
:
2010 if (member_vnfd
.vnfd_id_ref
== vnfd_msg
.id) and \
2011 (member_vnfd
.member_vnf_index_ref
== const_vnfd
.member_vnf_index
):
2012 group_info
= self
.resolve_placement_group_cloud_construct(group
)
2013 if group_info
is None:
2014 self
._log
.error("Could not resolve cloud-construct for placement group: %s", group
.name
)
2015 ### raise PlacementGroupError("Could not resolve cloud-construct for placement group: {}".format(group.name))
2017 self
._log
.info("Successfully resolved cloud construct for placement group: %s for VNF: %s (Member Index: %s)",
2020 const_vnfd
.member_vnf_index
)
2021 placement_groups
.append(group_info
)
2022 return placement_groups
2025 def create_vnf_record(self
, vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
, group_name
=None, group_instance_id
=None):
2026 # Fetch the VNFD associated with this VNF
2027 placement_groups
= self
.get_placement_groups(vnfd_msg
, const_vnfd
)
2028 self
._log
.info("Cloud Account for VNF %d is %s",const_vnfd
.member_vnf_index
,cloud_account_name
)
2029 self
._log
.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s",
2031 const_vnfd
.member_vnf_index
,
2032 [ group
.name
for group
in placement_groups
])
2033 vnfr
= yield from VirtualNetworkFunctionRecord
.create_record(self
._dts
,
2046 restart_mode
=self
.restart_mode
,
2048 if vnfr
.id in self
._vnfrs
:
2049 err
= "VNF with VNFR id %s already in vnf list" % (vnfr
.id,)
2050 raise NetworkServiceRecordError(err
)
2052 self
._vnfrs
[vnfr
.id] = vnfr
2053 self
._nsm
.vnfrs
[vnfr
.id] = vnfr
2055 yield from vnfr
.set_config_status(NsrYang
.ConfigStates
.INIT
)
2057 self
._log
.debug("Added VNFR %s to NSM VNFR list with id %s",
2063 def create_param_pools(self
):
2064 for param_pool
in self
.nsd_msg
.parameter_pool
:
2065 self
._log
.debug("Found parameter pool %s in nsr id %s", param_pool
, self
.id)
2067 start_value
= param_pool
.range.start_value
2068 end_value
= param_pool
.range.end_value
2069 if end_value
< start_value
:
2070 raise NetworkServiceRecordError(
2071 "Parameter pool %s has invalid range (start: {}, end: {})".format(
2072 start_value
, end_value
2076 self
._param
_pools
[param_pool
.name
] = config_value_pool
.ParameterValuePool(
2079 range(start_value
, end_value
)
2083 def fetch_vnfr(self
, vnfr_path
):
2084 """ Fetch VNFR record """
2086 self
._log
.debug("Fetching VNFR with key %s while instantiating %s",
2088 res_iter
= yield from self
._dts
.query_read(vnfr_path
, rwdts
.XactFlag
.MERGE
)
2090 for ent
in res_iter
:
2091 res
= yield from ent
2097 def instantiate_vnfs(self
, vnfrs
):
2099 This function instantiates VNFs for every VNF in this Network Service
2101 self
._log
.debug("Instantiating %u VNFs in NS %s", len(vnfrs
), self
.id)
2103 self
._log
.debug("Instantiating VNF: %s in NS %s", vnf
, self
.id)
2104 yield from self
.nsm_plugin
.instantiate_vnf(self
, vnf
)
2107 def instantiate_vnffgs(self
):
2109 This function instantiates VNFFGs for every VNFFG in this Network Service
2111 self
._log
.debug("Instantiating %u VNFFGs in NS %s",
2112 len(self
.nsd_msg
.vnffgd
), self
.id)
2113 for _
, vnfr
in self
.vnfrs
.items():
2114 while vnfr
.state
in [VnfRecordState
.INSTANTIATION_PENDING
, VnfRecordState
.INIT
]:
2115 self
._log
.debug("Received vnfr state for vnfr %s is %s; retrying",vnfr
.name
,vnfr
.state
)
2116 yield from asyncio
.sleep(2, loop
=self
._loop
)
2117 if vnfr
.state
== VnfRecordState
.ACTIVE
:
2118 self
._log
.debug("Received vnfr state for vnfr %s is %s ",vnfr
.name
,vnfr
.state
)
2121 self
._log
.debug("Received vnfr state for vnfr %s is %s; failing vnffg creation",vnfr
.name
,vnfr
.state
)
2122 self
._vnffgr
_state
= VnffgRecordState
.FAILED
2125 self
._log
.info("Waiting for 90 seconds for VMs to come up")
2126 yield from asyncio
.sleep(90, loop
=self
._loop
)
2127 self
._log
.info("Starting VNFFG orchestration")
2128 for vnffg
in self
._vnffgrs
.values():
2129 self
._log
.debug("Instantiating VNFFG: %s in NS %s", vnffg
, self
.id)
2130 yield from vnffg
.instantiate()
2133 def instantiate_scaling_instances(self
, config_xact
):
2134 """ Instantiate any default scaling instances in this Network Service """
2135 for group
in self
._scaling
_groups
.values():
2136 for i
in range(group
.min_instance_count
):
2137 self
._log
.debug("Instantiating %s default scaling instance %s", group
, i
)
2138 yield from self
.create_scale_group_instance(
2139 group
.name
, i
, config_xact
, is_default
=True
2142 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2143 if group_msg
.scaling_group_name_ref
!= group
.name
:
2146 for instance
in group_msg
.instance
:
2147 self
._log
.debug("Reloading %s scaling instance %s", group_msg
, instance
.id)
2148 yield from self
.create_scale_group_instance(
2149 group
.name
, instance
.id, config_xact
, is_default
=False
2152 def has_scaling_instances(self
):
2153 """ Return boolean indicating if the network service has default scaling groups """
2154 for group
in self
._scaling
_groups
.values():
2155 if group
.min_instance_count
> 0:
2158 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2159 if len(group_msg
.instance
) > 0:
2166 """ This function publishes this NSR """
2167 self
._nsr
_msg
= self
.create_msg()
2169 self
._log
.debug("Publishing the NSR with xpath %s and nsr %s",
2173 if self
._debug
_running
:
2174 self
._log
.debug("Publishing NSR in RUNNING state!")
2177 with self
._dts
.transaction() as xact
:
2178 yield from self
._nsm
.nsr_handler
.update(xact
, self
.nsr_xpath
, self
._nsr
_msg
)
2179 if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
:
2180 self
._debug
_running
= True
2183 def unpublish(self
, xact
):
2184 """ Unpublish this NSR object """
2185 self
._log
.debug("Unpublishing Network service id %s", self
.id)
2186 yield from self
._nsm
.nsr_handler
.delete(xact
, self
.nsr_xpath
)
2189 def nsr_xpath(self
):
2190 """ Returns the xpath associated with this NSR """
2192 "D,/nsr:ns-instance-opdata" +
2193 "/nsr:nsr[nsr:ns-instance-config-ref = '{}']"
2197 def xpath_from_nsr(nsr
):
2198 """ Returns the xpath associated with this NSR op data"""
2199 return (NetworkServiceRecord
.XPATH
+
2200 "[nsr:ns-instance-config-ref = '{}']").format(nsr
.id)
2203 def nsd_xpath(self
):
2204 """ Return NSD config xpath."""
2206 "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}']"
2207 ).format(self
.nsd_id
)
2210 def instantiate(self
, config_xact
):
2211 """"Instantiates a NetworkServiceRecord.
2213 This function instantiates a Network service
2214 which involves the following steps,
2216 * Instantiate every VL in NSD by sending create VLR request to DTS.
2217 * Instantiate every VNF in NSD by sending create VNF reuqest to DTS.
2218 * Publish the NSR details to DTS
2221 nsr: The NSR configuration request containing nsr-id and nsd
2222 config_xact: The configuration transaction which initiated the instatiation
2225 NetworkServiceRecordError if the NSR creation fails
2231 self
._log
.debug("Instantiating NS - %s xact - %s", self
, config_xact
)
2233 # Move the state to INIITALIZING
2234 self
.set_state(NetworkServiceRecordState
.INIT
)
2236 event_descr
= "Instantiation Request Received NSR Id:%s" % self
.id
2237 self
.record_event("instantiating", event_descr
)
2240 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
2243 # Update ref count if nsd present in catalog
2244 self
._nsm
.get_nsd_ref(self
.nsd_id
)
2246 except NetworkServiceDescriptorError
:
2247 # This could be an NSD not in the nsd-catalog
2250 # Merge any config and initial config primitive values
2251 self
.config_store
.merge_nsd_config(self
.nsd_msg
)
2252 self
._log
.debug("Merged NSD: {}".format(self
.nsd_msg
.as_dict()))
2254 event_descr
= "Fetched NSD with descriptor id %s" % self
.nsd_id
2255 self
.record_event("nsd-fetched", event_descr
)
2257 if self
._nsd
is None:
2258 msg
= "Failed to fetch NSD with nsd-id [%s] for nsr-id %s"
2259 self
._log
.debug(msg
, self
.nsd_id
, self
.id)
2260 raise NetworkServiceRecordError(self
)
2262 self
._log
.debug("Got nsd result %s", self
._nsd
)
2264 # Substitute any input parameters
2265 self
.substitute_input_parameters(self
._nsd
, self
._nsr
_cfg
_msg
)
2268 yield from self
.create(config_xact
)
2270 # Publish the NSR to DTS
2271 yield from self
.publish()
2274 def do_instantiate():
2276 Instantiate network service
2278 self
._log
.debug("Instantiating VLs nsr id [%s] nsd id [%s]",
2279 self
.id, self
.nsd_id
)
2281 # instantiate the VLs
2282 event_descr
= ("Instantiating %s external VLs for NSR id %s" %
2283 (len(self
.nsd_msg
.vld
), self
.id))
2284 self
.record_event("begin-external-vls-instantiation", event_descr
)
2286 self
.set_state(NetworkServiceRecordState
.VL_INIT_PHASE
)
2288 yield from self
.instantiate_vls()
2290 # Publish the NSR to DTS
2291 yield from self
.publish()
2293 event_descr
= ("Finished instantiating %s external VLs for NSR id %s" %
2294 (len(self
.nsd_msg
.vld
), self
.id))
2295 self
.record_event("end-external-vls-instantiation", event_descr
)
2297 self
.set_state(NetworkServiceRecordState
.VNF_INIT_PHASE
)
2299 self
._log
.debug("Instantiating VNFs ...... nsr[%s], nsd[%s]",
2300 self
.id, self
.nsd_id
)
2302 # instantiate the VNFs
2303 event_descr
= ("Instantiating %s VNFS for NSR id %s" %
2304 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2306 self
.record_event("begin-vnf-instantiation", event_descr
)
2308 yield from self
.instantiate_vnfs(self
._vnfrs
.values())
2310 self
._log
.debug(" Finished instantiating %d VNFs for NSR id %s",
2311 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
2313 event_descr
= ("Finished instantiating %s VNFs for NSR id %s" %
2314 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2315 self
.record_event("end-vnf-instantiation", event_descr
)
2317 if len(self
.vnffgrs
) > 0:
2318 #self.set_state(NetworkServiceRecordState.VNFFG_INIT_PHASE)
2319 event_descr
= ("Instantiating %s VNFFGS for NSR id %s" %
2320 (len(self
.nsd_msg
.vnffgd
), self
.id))
2322 self
.record_event("begin-vnffg-instantiation", event_descr
)
2324 yield from self
.instantiate_vnffgs()
2326 event_descr
= ("Finished instantiating %s VNFFGDs for NSR id %s" %
2327 (len(self
.nsd_msg
.vnffgd
), self
.id))
2328 self
.record_event("end-vnffg-instantiation", event_descr
)
2330 if self
.has_scaling_instances():
2331 event_descr
= ("Instantiating %s Scaling Groups for NSR id %s" %
2332 (len(self
._scaling
_groups
), self
.id))
2334 self
.record_event("begin-scaling-group-instantiation", event_descr
)
2335 yield from self
.instantiate_scaling_instances(config_xact
)
2336 self
.record_event("end-scaling-group-instantiation", event_descr
)
2338 # Give the plugin a chance to deploy the network service now that all
2339 # virtual links and vnfs are instantiated
2340 yield from self
.nsm_plugin
.deploy(self
._nsr
_msg
)
2342 self
._log
.debug("Publishing NSR...... nsr[%s], nsd[%s]",
2343 self
.id, self
.nsd_id
)
2345 # Publish the NSR to DTS
2346 yield from self
.publish()
2348 self
._log
.debug("Published NSR...... nsr[%s], nsd[%s]",
2349 self
.id, self
.nsd_id
)
2351 def on_instantiate_done(fut
):
2352 # If the do_instantiate fails, then publish NSR with failed result
2353 if fut
.exception() is not None:
2354 self
._log
.error("NSR instantiation failed for NSR id %s: %s", self
.id, str(fut
.exception()))
2355 self
._loop
.create_task(self
.instantiation_failed(failed_reason
=str(fut
.exception())))
2357 instantiate_task
= self
._loop
.create_task(do_instantiate())
2358 instantiate_task
.add_done_callback(on_instantiate_done
)
2361 def set_config_status(self
, status
, status_details
=None):
2362 if self
.config_status
!= status
:
2363 self
._log
.debug("Updating NSR {} status for {} to {}".
2364 format(self
.name
, self
.config_status
, status
))
2365 self
._config
_status
= status
2366 self
._config
_status
_details
= status_details
2368 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2369 self
.record_event("config-failed", "NS configuration failed",
2370 evt_details
=self
._config
_status
_details
)
2372 yield from self
.publish()
2375 def is_active(self
):
2376 """ This NS is active """
2377 self
.set_state(NetworkServiceRecordState
.RUNNING
)
2381 # Publish the NSR to DTS
2382 self
._log
.debug("Network service %s is active ", self
.id)
2383 self
._is
_active
= True
2385 event_descr
= "NSR in running state for NSR id %s" % self
.id
2386 self
.record_event("ns-running", event_descr
)
2388 yield from self
.publish()
2391 def instantiation_failed(self
, failed_reason
=None):
2392 """ The NS instantiation failed"""
2393 self
._log
.error("Network service id:%s, name:%s instantiation failed",
2395 self
.set_state(NetworkServiceRecordState
.FAILED
)
2397 event_descr
= "Instantiation of NS %s failed" % self
.id
2398 self
.record_event("ns-failed", event_descr
, evt_details
=failed_reason
)
2400 # Publish the NSR to DTS
2401 yield from self
.publish()
2404 def terminate_vnfrs(self
, vnfrs
):
2405 """ Terminate VNFRS in this network service """
2406 self
._log
.debug("Terminating VNFs in network service %s", self
.id)
2408 yield from self
.nsm_plugin
.terminate_vnf(vnfr
)
2411 def terminate(self
):
2412 """ Terminate a NetworkServiceRecord."""
2413 def terminate_vnffgrs():
2414 """ Terminate VNFFGRS in this network service """
2415 self
._log
.debug("Terminating VNFFGRs in network service %s", self
.id)
2416 for vnffgr
in self
.vnffgrs
.values():
2417 yield from vnffgr
.terminate()
2419 def terminate_vlrs():
2420 """ Terminate VLRs in this netork service """
2421 self
._log
.debug("Terminating VLs in network service %s", self
.id)
2422 for vlr
in self
.vlrs
:
2423 yield from self
.nsm_plugin
.terminate_vl(vlr
)
2424 vlr
.state
= VlRecordState
.TERMINATED
2425 if vlr
.id in self
.vlr_uptime_tasks
:
2426 self
.vlr_uptime_tasks
[vlr
.id].cancel()
2428 self
._log
.debug("Terminating network service id %s", self
.id)
2430 # Move the state to TERMINATE
2431 self
.set_state(NetworkServiceRecordState
.TERMINATE
)
2432 event_descr
= "Terminate being processed for NS Id:%s" % self
.id
2433 self
.record_event("terminate", event_descr
)
2435 # Move the state to VNF_TERMINATE_PHASE
2436 self
._log
.debug("Terminating VNFFGs in NS ID: %s", self
.id)
2437 self
.set_state(NetworkServiceRecordState
.VNFFG_TERMINATE_PHASE
)
2438 event_descr
= "Terminating VNFFGS in NS Id:%s" % self
.id
2439 self
.record_event("terminating-vnffgss", event_descr
)
2440 yield from terminate_vnffgrs()
2442 # Move the state to VNF_TERMINATE_PHASE
2443 self
.set_state(NetworkServiceRecordState
.VNF_TERMINATE_PHASE
)
2444 event_descr
= "Terminating VNFS in NS Id:%s" % self
.id
2445 self
.record_event("terminating-vnfs", event_descr
)
2446 yield from self
.terminate_vnfrs(self
.vnfrs
.values())
2448 # Move the state to VL_TERMINATE_PHASE
2449 self
.set_state(NetworkServiceRecordState
.VL_TERMINATE_PHASE
)
2450 event_descr
= "Terminating VLs in NS Id:%s" % self
.id
2451 self
.record_event("terminating-vls", event_descr
)
2452 yield from terminate_vlrs()
2454 yield from self
.nsm_plugin
.terminate_ns(self
)
2456 # Move the state to TERMINATED
2457 self
.set_state(NetworkServiceRecordState
.TERMINATED
)
2458 event_descr
= "Terminated NS Id:%s" % self
.id
2459 self
.record_event("terminated", event_descr
)
2462 """"Enable a NetworkServiceRecord."""
2466 """"Disable a NetworkServiceRecord."""
2469 def map_config_status(self
):
2470 self
._log
.debug("Config status for ns {} is {}".
2471 format(self
.name
, self
._config
_status
))
2472 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURING
:
2473 return 'configuring'
2474 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2478 def vl_phase_completed(self
):
2479 """ Are VLs created in this NS?"""
2480 return self
._vl
_phase
_completed
2482 def vnf_phase_completed(self
):
2483 """ Are VLs created in this NS?"""
2484 return self
._vnf
_phase
_completed
2486 def create_msg(self
):
2487 """ The network serice record as a message """
2488 nsr_dict
= {"ns_instance_config_ref": self
.id}
2489 nsr
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr
.from_dict(nsr_dict
)
2490 #nsr.cloud_account = self.cloud_account_name
2491 nsr
.sdn_account
= self
._sdn
_account
_name
2492 nsr
.name_ref
= self
.name
2493 nsr
.nsd_ref
= self
.nsd_id
2494 nsr
.nsd_name_ref
= self
.nsd_msg
.name
2495 nsr
.operational_events
= self
._op
_status
.msg
2496 nsr
.operational_status
= self
._op
_status
.yang_str()
2497 nsr
.config_status
= self
.map_config_status()
2498 nsr
.config_status_details
= self
._config
_status
_details
2499 nsr
.create_time
= self
._create
_time
2500 nsr
.uptime
= int(time
.time()) - self
._create
_time
2502 for cfg_prim
in self
.nsd_msg
.service_primitive
:
2503 cfg_prim
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_ServicePrimitive
.from_dict(
2505 nsr
.service_primitive
.append(cfg_prim
)
2507 for init_cfg
in self
.nsd_msg
.initial_config_primitive
:
2508 prim
= NsrYang
.NsrInitialConfigPrimitive
.from_dict(
2510 nsr
.initial_config_primitive
.append(prim
)
2512 if self
.vl_phase_completed():
2513 for vlr
in self
.vlrs
:
2514 nsr
.vlr
.append(vlr
.create_nsr_vlr_msg(self
.vnfrs
.values()))
2516 if self
.vnf_phase_completed():
2517 for vnfr_id
in self
.vnfrs
:
2518 nsr
.constituent_vnfr_ref
.append(self
.vnfrs
[vnfr_id
].const_vnfr_msg
)
2519 for vnffgr
in self
.vnffgrs
.values():
2520 nsr
.vnffgr
.append(vnffgr
.fetch_vnffgr())
2521 for scaling_group
in self
._scaling
_groups
.values():
2522 nsr
.scaling_group_record
.append(scaling_group
.create_record_msg())
2526 def all_vnfs_active(self
):
2527 """ Are all VNFS in this NS active? """
2528 for _
, vnfr
in self
.vnfrs
.items():
2529 if vnfr
.active
is not True:
2534 def update_state(self
):
2535 """ Re-evaluate this NS's state """
2536 curr_state
= self
._op
_status
.state
2538 if curr_state
== NetworkServiceRecordState
.TERMINATED
:
2539 self
._log
.debug("NS (%s) in terminated state, not updating state", self
.id)
2542 new_state
= NetworkServiceRecordState
.RUNNING
2543 self
._log
.info("Received update_state for nsr: %s, curr-state: %s",
2544 self
.id, curr_state
)
2546 # Check all the VNFRs are present
2547 for _
, vnfr
in self
.vnfrs
.items():
2548 if vnfr
.state
in [VnfRecordState
.ACTIVE
, VnfRecordState
.TERMINATED
]:
2550 elif vnfr
.state
== VnfRecordState
.FAILED
:
2551 if vnfr
._prev
_state
!= vnfr
.state
:
2552 event_descr
= "Instantiation of VNF %s failed" % vnfr
.id
2553 event_error_details
= vnfr
.state_failed_reason
2554 self
.record_event("vnf-failed", event_descr
, evt_details
=event_error_details
)
2555 vnfr
.set_state(VnfRecordState
.FAILED
)
2557 self
._log
.info("VNF state did not change, curr=%s, prev=%s",
2558 vnfr
.state
, vnfr
._prev
_state
)
2559 new_state
= NetworkServiceRecordState
.FAILED
2562 self
._log
.info("VNF %s in NSR %s is still not active; current state is: %s",
2563 vnfr
.id, self
.id, vnfr
.state
)
2564 new_state
= curr_state
2566 # If new state is RUNNING; check all VLs
2567 if new_state
== NetworkServiceRecordState
.RUNNING
:
2568 for vl
in self
.vlrs
:
2570 if vl
.state
in [VlRecordState
.ACTIVE
, VlRecordState
.TERMINATED
]:
2572 elif vl
.state
== VlRecordState
.FAILED
:
2573 if vl
.prev_state
!= vl
.state
:
2574 event_descr
= "Instantiation of VL %s failed" % vl
.id
2575 event_error_details
= vl
.state_failed_reason
2576 self
.record_event("vl-failed", event_descr
, evt_details
=event_error_details
)
2577 vl
.prev_state
= vl
.state
2579 self
._log
.debug("VL %s already in failed state")
2581 if vl
.state
in [VlRecordState
.INSTANTIATION_PENDING
, VlRecordState
.INIT
]:
2582 new_state
= NetworkServiceRecordState
.VL_INSTANTIATE
2585 if vl
.state
in [VlRecordState
.TERMINATE_PENDING
]:
2586 new_state
= NetworkServiceRecordState
.VL_TERMINATE
2589 # If new state is RUNNING; check VNFFGRs are also active
2590 if new_state
== NetworkServiceRecordState
.RUNNING
:
2591 for _
, vnffgr
in self
.vnffgrs
.items():
2592 self
._log
.info("Checking vnffgr state for nsr %s is: %s",
2593 self
.id, vnffgr
.state
)
2594 if vnffgr
.state
== VnffgRecordState
.ACTIVE
:
2596 elif vnffgr
.state
== VnffgRecordState
.FAILED
:
2597 event_descr
= "Instantiation of VNFFGR %s failed" % vnffgr
.id
2598 self
.record_event("vnffg-failed", event_descr
)
2599 new_state
= NetworkServiceRecordState
.FAILED
2602 self
._log
.info("VNFFGR %s in NSR %s is still not active; current state is: %s",
2603 vnffgr
.id, self
.id, vnffgr
.state
)
2604 new_state
= curr_state
2606 # Update all the scaling group instance operational status to
2607 # reflect the state of all VNFR within that instance
2608 yield from self
._update
_scale
_group
_instances
_status
()
2610 for _
, group
in self
._scaling
_groups
.items():
2611 if group
.state
== scale_group
.ScaleGroupState
.SCALING_OUT
:
2612 new_state
= NetworkServiceRecordState
.SCALING_OUT
2614 elif group
.state
== scale_group
.ScaleGroupState
.SCALING_IN
:
2615 new_state
= NetworkServiceRecordState
.SCALING_IN
2618 if new_state
!= curr_state
:
2619 self
._log
.debug("Changing state of Network service %s from %s to %s",
2620 self
.id, curr_state
, new_state
)
2621 if new_state
== NetworkServiceRecordState
.RUNNING
:
2622 yield from self
.is_active()
2623 elif new_state
== NetworkServiceRecordState
.FAILED
:
2624 # If the NS is already active and we entered scaling_in, scaling_out,
2625 # do not mark the NS as failing if scaling operation failed.
2626 if curr_state
in [NetworkServiceRecordState
.SCALING_OUT
,
2627 NetworkServiceRecordState
.SCALING_IN
] and self
._is
_active
:
2628 new_state
= NetworkServiceRecordState
.RUNNING
2629 self
.set_state(new_state
)
2631 yield from self
.instantiation_failed()
2633 self
.set_state(new_state
)
2635 yield from self
.publish()
2638 class InputParameterSubstitution(object):
2640 This class is responsible for substituting input parameters into an NSD.
2643 def __init__(self
, log
):
2644 """Create an instance of InputParameterSubstitution
2647 log - a logger for this object to use
2652 def __call__(self
, nsd
, nsr_config
):
2653 """Substitutes input parameters from the NSR config into the NSD
2655 This call modifies the provided NSD with the input parameters that are
2656 contained in the NSR config.
2659 nsd - a GI NSD object
2660 nsr_config - a GI NSR config object
2663 if nsd
is None or nsr_config
is None:
2666 # Create a lookup of the xpath elements that this descriptor allows
2668 optional_input_parameters
= set()
2669 for input_parameter
in nsd
.input_parameter_xpath
:
2670 optional_input_parameters
.add(input_parameter
.xpath
)
2672 # Apply the input parameters to the descriptor
2673 if nsr_config
.input_parameter
:
2674 for param
in nsr_config
.input_parameter
:
2675 if param
.xpath
not in optional_input_parameters
:
2676 msg
= "tried to set an invalid input parameter ({})"
2677 self
.log
.error(msg
.format(param
.xpath
))
2681 "input-parameter:{} = {}".format(
2688 xpath
.setxattr(nsd
, param
.xpath
, param
.value
)
2690 except Exception as e
:
2691 self
.log
.exception(e
)
2694 class NetworkServiceDescriptor(object):
2696 Network service descriptor class
2699 def __init__(self
, dts
, log
, loop
, nsd
, nsm
):
2711 """ Returns nsd id """
2716 """ Returns name of nsd """
2717 return self
._nsd
.name
2720 def ref_count(self
):
2721 """ Returns reference count"""
2722 return self
._ref
_count
2725 """ Returns whether nsd is in use or not """
2726 return True if self
.ref_count
> 0 else False
2729 """ Take a reference on this object """
2730 self
._ref
_count
+= 1
2733 """ Release reference on this object """
2734 if self
.ref_count
< 1:
2735 msg
= ("Unref on a NSD object - nsd id %s, ref_count = %s" %
2736 (self
.id, self
.ref_count
))
2737 self
._log
.critical(msg
)
2738 raise NetworkServiceDescriptorError(msg
)
2739 self
._ref
_count
-= 1
2743 """ Return the message associated with this NetworkServiceDescriptor"""
2747 def path_for_id(nsd_id
):
2748 """ Return path for the passed nsd_id"""
2749 return "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}'".format(nsd_id
)
2752 """ Return the message associated with this NetworkServiceDescriptor"""
2753 return NetworkServiceDescriptor
.path_for_id(self
.id)
2755 def update(self
, nsd
):
2756 """ Update the NSD descriptor """
2760 class NsdDtsHandler(object):
2761 """ The network service descriptor DTS handler """
2762 XPATH
= "C,/nsd:nsd-catalog/nsd:nsd"
2764 def __init__(self
, dts
, log
, loop
, nsm
):
2774 """ Return registration handle """
2779 """ Register for Nsd create/update/delete/read requests from dts """
2781 def on_apply(dts
, acg
, xact
, action
, scratch
):
2782 """Apply the configuration"""
2783 is_recovery
= xact
.xact
is None and action
== rwdts
.AppconfAction
.INSTALL
2784 self
._log
.debug("Got nsd apply cfg (xact:%s) (action:%s)",
2786 # Create/Update an NSD record
2787 for cfg
in self
._regh
.get_xact_elements(xact
):
2788 # Only interested in those NSD cfgs whose ID was received in prepare callback
2789 if cfg
.id in scratch
.get('nsds', []) or is_recovery
:
2790 self
._nsm
.update_nsd(cfg
)
2792 scratch
.pop('nsds', None)
2794 return RwTypes
.RwStatus
.SUCCESS
2797 def delete_nsd_libs(nsd_id
):
2798 """ Remove any files uploaded with NSD and stored under $RIFT_ARTIFACTS/libs/<id> """
2800 rift_artifacts_dir
= os
.environ
['RIFT_ARTIFACTS']
2801 nsd_dir
= os
.path
.join(rift_artifacts_dir
, 'launchpad/libs', nsd_id
)
2803 if os
.path
.exists (nsd_dir
):
2804 shutil
.rmtree(nsd_dir
, ignore_errors
=True)
2805 except Exception as e
:
2806 self
._log
.error("Exception in cleaning up NSD libs {}: {}".
2808 self
._log
.excpetion(e
)
2811 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2812 """ Prepare callback from DTS for NSD config """
2814 self
._log
.info("Got nsd prepare - config received nsd id %s, msg %s",
2817 fref
= ProtobufC
.FieldReference
.alloc()
2818 fref
.goto_whole_message(msg
.to_pbcm())
2820 if fref
.is_field_deleted():
2821 # Delete an NSD record
2822 self
._log
.debug("Deleting NSD with id %s", msg
.id)
2823 if self
._nsm
.nsd_in_use(msg
.id):
2824 self
._log
.debug("Cannot delete NSD in use - %s", msg
.id)
2825 err
= "Cannot delete an NSD in use - %s" % msg
.id
2826 raise NetworkServiceDescriptorRefCountExists(err
)
2828 yield from delete_nsd_libs(msg
.id)
2829 self
._nsm
.delete_nsd(msg
.id)
2831 # Add this NSD to scratch to create/update in apply callback
2832 nsds
= scratch
.setdefault('nsds', [])
2834 # acg._scratch['nsds'].append(msg.id)
2836 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2839 "Registering for NSD config using xpath: %s",
2840 NsdDtsHandler
.XPATH
,
2843 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2844 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2845 # Need a list in scratch to store NSDs to create/update later
2846 # acg._scratch['nsds'] = list()
2847 self
._regh
= acg
.register(
2848 xpath
=NsdDtsHandler
.XPATH
,
2849 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
2850 on_prepare
=on_prepare
)
2853 class VnfdDtsHandler(object):
2854 """ DTS handler for VNFD config changes """
2855 XPATH
= "C,/vnfd:vnfd-catalog/vnfd:vnfd"
2857 def __init__(self
, dts
, log
, loop
, nsm
):
2866 """ DTS registration handle """
2871 """ Register for VNFD configuration"""
2874 def on_apply(dts
, acg
, xact
, action
, scratch
):
2875 """Apply the configuration"""
2876 self
._log
.debug("Got NSM VNFD apply (xact: %s) (action: %s)(scr: %s)",
2877 xact
, action
, scratch
)
2879 # Create/Update a VNFD record
2880 for cfg
in self
._regh
.get_xact_elements(xact
):
2881 # Only interested in those VNFD cfgs whose ID was received in prepare callback
2882 if cfg
.id in scratch
.get('vnfds', []):
2883 self
._nsm
.update_vnfd(cfg
)
2885 for cfg
in self
._regh
.elements
:
2886 if cfg
.id in scratch
.get('deleted_vnfds', []):
2887 yield from self
._nsm
.delete_vnfd(cfg
.id)
2889 scratch
.pop('vnfds', None)
2890 scratch
.pop('deleted_vnfds', None)
2893 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2894 """ on prepare callback """
2895 self
._log
.debug("Got on prepare for VNFD (path: %s) (action: %s) (msg: %s)",
2896 ks_path
.to_xpath(RwNsmYang
.get_schema()), xact_info
.query_action
, msg
)
2898 fref
= ProtobufC
.FieldReference
.alloc()
2899 fref
.goto_whole_message(msg
.to_pbcm())
2901 # Handle deletes in prepare_callback, but adds/updates in apply_callback
2902 if fref
.is_field_deleted():
2903 self
._log
.debug("Adding msg to deleted field")
2904 deleted_vnfds
= scratch
.setdefault('deleted_vnfds', [])
2905 deleted_vnfds
.append(msg
.id)
2907 # Add this VNFD to scratch to create/update in apply callback
2908 vnfds
= scratch
.setdefault('vnfds', [])
2909 vnfds
.append(msg
.id)
2911 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2914 "Registering for VNFD config using xpath: %s",
2915 VnfdDtsHandler
.XPATH
,
2917 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2918 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2919 # Need a list in scratch to store VNFDs to create/update later
2920 # acg._scratch['vnfds'] = list()
2921 # acg._scratch['deleted_vnfds'] = list()
2922 self
._regh
= acg
.register(
2923 xpath
=VnfdDtsHandler
.XPATH
,
2924 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY
,
2925 on_prepare
=on_prepare
)
2927 class NsrRpcDtsHandler(object):
2928 """ The network service instantiation RPC DTS handler """
2929 EXEC_NSR_CONF_XPATH
= "I,/nsr:start-network-service"
2930 EXEC_NSR_CONF_O_XPATH
= "O,/nsr:start-network-service"
2931 NETCONF_IP_ADDRESS
= "127.0.0.1"
2933 RESTCONF_PORT
= 8888
2934 NETCONF_USER
= "admin"
2935 NETCONF_PW
= "admin"
2936 REST_BASE_V2_URL
= 'https://{}:{}/v2/api/'.format("127.0.0.1",8888)
2938 def __init__(self
, dts
, log
, loop
, nsm
):
2945 self
._ns
_regh
= None
2947 self
._manager
= None
2948 self
._nsr
_config
_url
= NsrRpcDtsHandler
.REST_BASE_V2_URL
+ 'config/ns-instance-config'
2950 self
._model
= RwYang
.Model
.create_libncx()
2951 self
._model
.load_schema_ypbc(RwNsrYang
.get_schema())
2955 """ Return the NS manager instance """
2959 def wrap_netconf_config_xml(xml
):
2960 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
2964 def _connect(self
, timeout_secs
=240):
2966 start_time
= time
.time()
2967 while (time
.time() - start_time
) < timeout_secs
:
2970 self
._log
.debug("Attemping NsmTasklet netconf connection.")
2972 manager
= yield from ncclient
.asyncio_manager
.asyncio_connect(
2974 host
=NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
,
2975 port
=NsrRpcDtsHandler
.NETCONF_PORT
,
2976 username
=NsrRpcDtsHandler
.NETCONF_USER
,
2977 password
=NsrRpcDtsHandler
.NETCONF_PW
,
2979 look_for_keys
=False,
2980 hostkey_verify
=False,
2985 except ncclient
.transport
.errors
.SSHError
as e
:
2986 self
._log
.warning("Netconf connection to launchpad %s failed: %s",
2987 NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
, str(e
))
2989 yield from asyncio
.sleep(5, loop
=self
._loop
)
2991 raise NsrInstantiationFailed("Failed to connect to Launchpad within %s seconds" %
2994 def _apply_ns_instance_config(self
,payload_dict
):
2995 #self._log.debug("At apply NS instance config with payload %s",payload_dict)
2996 req_hdr
= {'accept':'application/vnd.yang.data+json','content-type':'application/vnd.yang.data+json'}
2997 response
=requests
.post(self
._nsr
_config
_url
, headers
=req_hdr
, auth
=('admin', 'admin'),data
=payload_dict
,verify
=False)
3002 """ Register for NS monitoring read from dts """
3004 def on_ns_config_prepare(xact_info
, action
, ks_path
, msg
):
3005 """ prepare callback from dts start-network-service"""
3006 assert action
== rwdts
.QueryAction
.RPC
3008 rpc_op
= NsrYang
.YangOutput_Nsr_StartNetworkService
.from_dict({
3009 "nsr_id":str(uuid
.uuid4())
3012 if not ('name' in rpc_ip
and 'nsd_ref' in rpc_ip
and ('cloud_account' in rpc_ip
or 'om_datacenter' in rpc_ip
)):
3013 self
._log
.error("Mandatory parameters name or nsd_ref or cloud account not found in start-network-service {}".format(rpc_ip
))
3016 self
._log
.debug("start-network-service RPC input: {}".format(rpc_ip
))
3019 # Add used value to the pool
3020 self
._log
.debug("RPC output: {}".format(rpc_op
))
3022 nsd_copy
= self
.nsm
.get_nsd(rpc_ip
.nsd_ref
)
3024 #if not self._manager:
3025 # self._manager = yield from self._connect()
3027 self
._log
.debug("Configuring ns-instance-config with name %s nsd-ref: %s",
3028 rpc_ip
.name
, rpc_ip
.nsd_ref
)
3030 ns_instance_config_dict
= {"id":rpc_op
.nsr_id
, "admin_status":"ENABLED"}
3031 ns_instance_config_copy_dict
= {k
:v
for k
, v
in rpc_ip
.as_dict().items()
3032 if k
in RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr().fields
}
3033 ns_instance_config_dict
.update(ns_instance_config_copy_dict
)
3035 ns_instance_config
= RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.from_dict(ns_instance_config_dict
)
3036 ns_instance_config
.nsd
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_Nsd()
3037 ns_instance_config
.nsd
.from_dict(nsd_copy
.msg
.as_dict())
3039 payload_dict
= ns_instance_config
.to_json(self
._model
)
3040 #xml = ns_instance_config.to_xml_v2(self._model)
3041 #netconf_xml = self.wrap_netconf_config_xml(xml)
3043 #self._log.debug("Sending configure ns-instance-config xml to %s: %s",
3044 # netconf_xml, NsrRpcDtsHandler.NETCONF_IP_ADDRESS)
3045 self
._log
.debug("Sending configure ns-instance-config json to %s: %s",
3046 self
._nsr
_config
_url
,ns_instance_config
)
3048 #response = yield from self._manager.edit_config(
3050 # config=netconf_xml,
3052 response
= yield from self
._loop
.run_in_executor(
3054 self
._apply
_ns
_instance
_config
,
3057 response
.raise_for_status()
3058 self
._log
.debug("Received edit config response: %s", response
.json())
3060 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
,
3061 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
,
3063 except Exception as e
:
3064 self
._log
.error("Exception processing the "
3065 "start-network-service: {}".format(e
))
3066 self
._log
.exception(e
)
3067 xact_info
.respond_xpath(rwdts
.XactRspCode
.NACK
,
3068 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
)
3071 hdl_ns
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_ns_config_prepare
,)
3073 with self
._dts
.group_create() as group
:
3074 self
._ns
_regh
= group
.register(xpath
=NsrRpcDtsHandler
.EXEC_NSR_CONF_XPATH
,
3076 flags
=rwdts
.Flag
.PUBLISHER
,
3080 class NsrDtsHandler(object):
3081 """ The network service DTS handler """
3082 NSR_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr"
3083 SCALE_INSTANCE_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr/nsr:scaling-group/nsr:instance"
3084 KEY_PAIR_XPATH
= "C,/nsr:key-pair"
3086 def __init__(self
, dts
, log
, loop
, nsm
):
3092 self
._nsr
_regh
= None
3093 self
._scale
_regh
= None
3094 self
._key
_pair
_regh
= None
3098 """ Return the NS manager instance """
3103 """ Register for Nsr create/update/delete/read requests from dts """
3105 def nsr_id_from_keyspec(ks
):
3106 nsr_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.schema().keyspec_to_entry(ks
)
3107 nsr_id
= nsr_path_entry
.key00
.id
3110 def group_name_from_keyspec(ks
):
3111 group_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
.schema().keyspec_to_entry(ks
)
3112 group_name
= group_path_entry
.key00
.scaling_group_name_ref
3115 def is_instance_in_reg_elements(nsr_id
, group_name
, instance_id
):
3116 """ Return boolean indicating if scaling group instance was already commited previously.
3118 By looking at the existing elements in this registration handle (elements not part
3119 of this current xact), we can tell if the instance was configured previously without
3120 keeping any application state.
3122 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3123 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3124 elem_group_name
= group_name_from_keyspec(keyspec
)
3126 if elem_nsr_id
!= nsr_id
or group_name
!= elem_group_name
:
3129 if instance_cfg
.id == instance_id
:
3134 def get_scale_group_instance_delta(nsr_id
, group_name
, xact
):
3135 delta
= {"added": [], "deleted": []}
3136 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3137 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3138 if elem_nsr_id
!= nsr_id
:
3141 elem_group_name
= group_name_from_keyspec(keyspec
)
3142 if elem_group_name
!= group_name
:
3145 delta
["added"].append(instance_cfg
.id)
3147 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(include_keyspec
=True):
3148 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3149 if elem_nsr_id
!= nsr_id
:
3152 elem_group_name
= group_name_from_keyspec(keyspec
)
3153 if elem_group_name
!= group_name
:
3156 if instance_cfg
.id in delta
["added"]:
3157 delta
["added"].remove(instance_cfg
.id)
3159 delta
["deleted"].append(instance_cfg
.id)
3164 def update_nsr_nsd(nsr_id
, xact
, scratch
):
3167 def get_nsr_vl_delta(nsr_id
, xact
, scratch
):
3168 delta
= {"added": [], "deleted": []}
3169 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3170 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3171 if elem_nsr_id
!= nsr_id
:
3174 if 'vld' in instance_cfg
.nsd
:
3175 for vld
in instance_cfg
.nsd
.vld
:
3176 delta
["added"].append(vld
)
3178 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3179 self
._log
.debug("NSR update: %s", instance_cfg
)
3180 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3181 if elem_nsr_id
!= nsr_id
:
3184 if 'vld' in instance_cfg
.nsd
:
3185 for vld
in instance_cfg
.nsd
.vld
:
3186 if vld
in delta
["added"]:
3187 delta
["added"].remove(vld
)
3189 delta
["deleted"].append(vld
)
3193 vl_delta
= yield from get_nsr_vl_delta(nsr_id
, xact
, scratch
)
3194 self
._log
.debug("Got NSR:%s VL instance delta: %s", nsr_id
, vl_delta
)
3196 for vld
in vl_delta
["added"]:
3197 yield from self
._nsm
.nsr_instantiate_vl(nsr_id
, vld
)
3199 for vld
in vl_delta
["deleted"]:
3200 yield from self
._nsm
.nsr_terminate_vl(nsr_id
, vld
)
3202 def get_add_delete_update_cfgs(dts_member_reg
, xact
, key_name
, scratch
):
3203 # Unfortunately, it is currently difficult to figure out what has exactly
3204 # changed in this xact without Pbdelta support (RIFT-4916)
3205 # As a workaround, we can fetch the pre and post xact elements and
3206 # perform a comparison to figure out adds/deletes/updates
3207 xact_cfgs
= list(dts_member_reg
.get_xact_elements(xact
))
3208 curr_cfgs
= list(dts_member_reg
.elements
)
3210 xact_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in xact_cfgs
}
3211 curr_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in curr_cfgs
}
3214 added_keys
= set(xact_key_map
) - set(curr_key_map
)
3215 added_cfgs
= [xact_key_map
[key
] for key
in added_keys
]
3218 deleted_keys
= set(curr_key_map
) - set(xact_key_map
)
3219 deleted_cfgs
= [curr_key_map
[key
] for key
in deleted_keys
]
3222 updated_keys
= set(curr_key_map
) & set(xact_key_map
)
3223 updated_cfgs
= [xact_key_map
[key
] for key
in updated_keys
3224 if xact_key_map
[key
] != curr_key_map
[key
]]
3226 return added_cfgs
, deleted_cfgs
, updated_cfgs
3228 def get_nsr_key_pairs(dts_member_reg
, xact
):
3230 for instance_cfg
, keyspec
in dts_member_reg
.get_xact_elements(xact
, include_keyspec
=True):
3231 self
._log
.debug("Key pair received is {} KS: {}".format(instance_cfg
, keyspec
))
3232 xpath
= keyspec
.to_xpath(RwNsrYang
.get_schema())
3233 key_pairs
[instance_cfg
.name
] = instance_cfg
3236 def on_apply(dts
, acg
, xact
, action
, scratch
):
3237 """Apply the configuration"""
3238 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3239 xact
, action
, scratch
)
3241 def handle_create_nsr(msg
, key_pairs
=None, restart_mode
=False):
3242 # Handle create nsr requests """
3243 # Do some validations
3244 if not msg
.has_field("nsd"):
3245 err
= "NSD not provided"
3246 self
._log
.error(err
)
3247 raise NetworkServiceRecordError(err
)
3249 self
._log
.debug("Creating NetworkServiceRecord %s from nsr config %s",
3250 msg
.id, msg
.as_dict())
3251 nsr
= self
.nsm
.create_nsr(msg
, key_pairs
=key_pairs
, restart_mode
=restart_mode
)
3254 def handle_delete_nsr(msg
):
3256 def delete_instantiation(ns_id
):
3257 """ Delete instantiation """
3258 with self
._dts
.transaction() as xact
:
3259 yield from self
._nsm
.terminate_ns(ns_id
, xact
)
3261 # Handle delete NSR requests
3262 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3263 # Terminate the NSR instance
3264 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3266 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3267 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3268 nsr
.record_event("terminate-rcvd", event_descr
)
3270 self
._loop
.create_task(delete_instantiation(msg
.id))
3273 def begin_instantiation(nsr
):
3274 # Begin instantiation
3275 self
._log
.info("Beginning NS instantiation: %s", nsr
.id)
3276 yield from self
._nsm
.instantiate_ns(nsr
.id, xact
)
3278 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3279 xact
, action
, scratch
)
3281 if action
== rwdts
.AppconfAction
.INSTALL
and xact
.id is None:
3283 for element
in self
._key
_pair
_regh
.elements
:
3284 key_pairs
.append(element
)
3285 for element
in self
._nsr
_regh
.elements
:
3286 nsr
= handle_create_nsr(element
, key_pairs
, restart_mode
=True)
3287 self
._loop
.create_task(begin_instantiation(nsr
))
3290 (added_msgs
, deleted_msgs
, updated_msgs
) = get_add_delete_update_cfgs(self
._nsr
_regh
,
3294 self
._log
.debug("Added: %s, Deleted: %s, Updated: %s", added_msgs
,
3295 deleted_msgs
, updated_msgs
)
3297 for msg
in added_msgs
:
3298 if msg
.id not in self
._nsm
.nsrs
:
3299 self
._log
.info("Create NSR received in on_apply to instantiate NS:%s", msg
.id)
3300 key_pairs
= get_nsr_key_pairs(self
._key
_pair
_regh
, xact
)
3301 nsr
= handle_create_nsr(msg
,key_pairs
)
3302 self
._loop
.create_task(begin_instantiation(nsr
))
3304 for msg
in deleted_msgs
:
3305 self
._log
.info("Delete NSR received in on_apply to terminate NS:%s", msg
.id)
3307 handle_delete_nsr(msg
)
3309 self
._log
.exception("Failed to terminate NS:%s", msg
.id)
3311 for msg
in updated_msgs
:
3312 self
._log
.info("Update NSR received in on_apply: %s", msg
)
3314 self
._nsm
.nsr_update_cfg(msg
.id, msg
)
3317 self
._loop
.create_task(update_nsr_nsd(msg
.id, xact
, scratch
))
3319 for group
in msg
.scaling_group
:
3320 instance_delta
= get_scale_group_instance_delta(msg
.id, group
.scaling_group_name_ref
, xact
)
3321 self
._log
.debug("Got NSR:%s scale group instance delta: %s", msg
.id, instance_delta
)
3323 for instance_id
in instance_delta
["added"]:
3324 self
._nsm
.scale_nsr_out(msg
.id, group
.scaling_group_name_ref
, instance_id
, xact
)
3326 for instance_id
in instance_delta
["deleted"]:
3327 self
._nsm
.scale_nsr_in(msg
.id, group
.scaling_group_name_ref
, instance_id
)
3330 return RwTypes
.RwStatus
.SUCCESS
3333 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
3334 """ Prepare calllback from DTS for NSR """
3336 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3337 action
= xact_info
.query_action
3339 "Got Nsr prepare callback (xact: %s) (action: %s) (info: %s), %s:%s)",
3340 xact
, action
, xact_info
, xpath
, msg
3344 def delete_instantiation(ns_id
):
3345 """ Delete instantiation """
3346 yield from self
._nsm
.terminate_ns(ns_id
, None)
3348 def handle_delete_nsr():
3349 """ Handle delete NSR requests """
3350 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3351 # Terminate the NSR instance
3352 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3354 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3355 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3356 nsr
.record_event("terminate-rcvd", event_descr
)
3358 self
._loop
.create_task(delete_instantiation(msg
.id))
3360 fref
= ProtobufC
.FieldReference
.alloc()
3361 fref
.goto_whole_message(msg
.to_pbcm())
3363 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
, rwdts
.QueryAction
.DELETE
]:
3364 # if this is an NSR create
3365 if action
!= rwdts
.QueryAction
.DELETE
and msg
.id not in self
._nsm
.nsrs
:
3366 # Ensure the Cloud account/datacenter has been specified
3367 if not msg
.has_field("cloud_account") and not msg
.has_field("om_datacenter"):
3368 raise NsrInstantiationFailed("Cloud account or datacenter not specified in NSR")
3370 # Check if nsd is specified
3371 if not msg
.has_field("nsd"):
3372 raise NsrInstantiationFailed("NSD not specified in NSR")
3375 nsr
= self
._nsm
.nsrs
[msg
.id]
3377 if msg
.has_field("nsd"):
3378 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3379 raise NsrVlUpdateError("Unable to update VL when NSR not in running state")
3380 if 'vld' not in msg
.nsd
or len(msg
.nsd
.vld
) == 0:
3381 raise NsrVlUpdateError("NS config NSD should have atleast 1 VLD defined")
3383 if msg
.has_field("scaling_group"):
3384 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3385 raise ScalingOperationError("Unable to perform scaling action when NS is not in running state")
3387 if len(msg
.scaling_group
) > 1:
3388 raise ScalingOperationError("Only a single scaling group can be configured at a time")
3390 for group_msg
in msg
.scaling_group
:
3391 num_new_group_instances
= len(group_msg
.instance
)
3392 if num_new_group_instances
> 1:
3393 raise ScalingOperationError("Only a single scaling instance can be modified at a time")
3395 elif num_new_group_instances
== 1:
3396 scale_group
= nsr
.scaling_groups
[group_msg
.scaling_group_name_ref
]
3397 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
]:
3398 if len(scale_group
.instances
) == scale_group
.max_instance_count
:
3399 raise ScalingOperationError("Max instances for %s reached" % scale_group
)
3401 acg
.handle
.prepare_complete_ok(xact_info
.handle
)
3404 self
._log
.debug("Registering for NSR config using xpath: %s",
3405 NsrDtsHandler
.NSR_XPATH
)
3407 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
3408 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
3409 self
._nsr
_regh
= acg
.register(xpath
=NsrDtsHandler
.NSR_XPATH
,
3410 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3411 on_prepare
=on_prepare
)
3413 self
._scale
_regh
= acg
.register(
3414 xpath
=NsrDtsHandler
.SCALE_INSTANCE_XPATH
,
3415 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY| rwdts
.Flag
.CACHE
,
3418 self
._key
_pair
_regh
= acg
.register(
3419 xpath
=NsrDtsHandler
.KEY_PAIR_XPATH
,
3420 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3424 class NsrOpDataDtsHandler(object):
3425 """ The network service op data DTS handler """
3426 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
3428 def __init__(self
, dts
, log
, loop
, nsm
):
3437 """ Return the registration handle"""
3442 """ Return the NS manager instance """
3447 """ Register for Nsr op data publisher registration"""
3448 self
._log
.debug("Registering Nsr op data path %s as publisher",
3449 NsrOpDataDtsHandler
.XPATH
)
3451 hdl
= rift
.tasklets
.DTS
.RegistrationHandler()
3452 handlers
= rift
.tasklets
.Group
.Handler()
3453 with self
._dts
.group_create(handler
=handlers
) as group
:
3454 self
._regh
= group
.register(xpath
=NsrOpDataDtsHandler
.XPATH
,
3456 flags
=rwdts
.Flag
.PUBLISHER | rwdts
.Flag
.NO_PREP_READ | rwdts
.Flag
.DATASTORE
)
3459 def create(self
, path
, msg
):
3461 Create an NS record in DTS with the path and message
3463 self
._log
.debug("Creating NSR %s:%s", path
, msg
)
3464 self
.regh
.create_element(path
, msg
)
3465 self
._log
.debug("Created NSR, %s:%s", path
, msg
)
3468 def update(self
, path
, msg
, flags
=rwdts
.XactFlag
.REPLACE
):
3470 Update an NS record in DTS with the path and message
3472 self
._log
.debug("Updating NSR, %s:%s regh = %s", path
, msg
, self
.regh
)
3473 self
.regh
.update_element(path
, msg
, flags
)
3474 self
._log
.debug("Updated NSR, %s:%s", path
, msg
)
3477 def delete(self
, path
):
3479 Update an NS record in DTS with the path and message
3481 self
._log
.debug("Deleting NSR path:%s", path
)
3482 self
.regh
.delete_element(path
)
3483 self
._log
.debug("Deleted NSR path:%s", path
)
3486 class VnfrDtsHandler(object):
3487 """ The virtual network service DTS handler """
3488 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
3490 def __init__(self
, dts
, log
, loop
, nsm
):
3500 """ Return registration handle """
3505 """ Return the NS manager instance """
3510 """ Register for vnfr create/update/delete/ advises from dts """
3512 def on_commit(xact_info
):
3513 """ The transaction has been committed """
3514 self
._log
.debug("Got vnfr commit (xact_info: %s)", xact_info
)
3515 return rwdts
.MemberRspCode
.ACTION_OK
3518 def on_prepare(xact_info
, action
, ks_path
, msg
):
3519 """ prepare callback from dts """
3520 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3522 "Got vnfr on_prepare cb (xact_info: %s, action: %s): %s:%s",
3523 xact_info
, action
, ks_path
, msg
3526 schema
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr
.schema()
3527 path_entry
= schema
.keyspec_to_entry(ks_path
)
3528 if path_entry
.key00
.id not in self
._nsm
._vnfrs
:
3529 self
._log
.error("%s request for non existent record path %s",
3531 xact_info
.respond_xpath(rwdts
.XactRspCode
.NA
, xpath
)
3535 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3536 if action
== rwdts
.QueryAction
.CREATE
or action
== rwdts
.QueryAction
.UPDATE
:
3537 yield from self
._nsm
.update_vnfr(msg
)
3538 elif action
== rwdts
.QueryAction
.DELETE
:
3539 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3540 self
._nsm
.delete_vnfr(path_entry
.key00
.id)
3542 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
, xpath
)
3544 self
._log
.debug("Registering for VNFR using xpath: %s",
3545 VnfrDtsHandler
.XPATH
,)
3547 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_commit
=on_commit
,
3548 on_prepare
=on_prepare
,)
3549 with self
._dts
.group_create() as group
:
3550 self
._regh
= group
.register(xpath
=VnfrDtsHandler
.XPATH
,
3552 flags
=(rwdts
.Flag
.SUBSCRIBER
),)
3555 class NsdRefCountDtsHandler(object):
3556 """ The NSD Ref Count DTS handler """
3557 XPATH
= "D,/nsr:ns-instance-opdata/rw-nsr:nsd-ref-count"
3559 def __init__(self
, dts
, log
, loop
, nsm
):
3569 """ Return registration handle """
3574 """ Return the NS manager instance """
3579 """ Register for NSD ref count read from dts """
3582 def on_prepare(xact_info
, action
, ks_path
, msg
):
3583 """ prepare callback from dts """
3584 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3586 if action
== rwdts
.QueryAction
.READ
:
3587 schema
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount
.schema()
3588 path_entry
= schema
.keyspec_to_entry(ks_path
)
3589 nsd_list
= yield from self
._nsm
.get_nsd_refcount(path_entry
.key00
.nsd_id_ref
)
3590 for xpath
, msg
in nsd_list
:
3591 xact_info
.respond_xpath(rsp_code
=rwdts
.XactRspCode
.MORE
,
3594 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
3596 raise NetworkServiceRecordError("Not supported operation %s" % action
)
3598 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_prepare
,)
3599 with self
._dts
.group_create() as group
:
3600 self
._regh
= group
.register(xpath
=NsdRefCountDtsHandler
.XPATH
,
3602 flags
=rwdts
.Flag
.PUBLISHER
,)
3605 class NsManager(object):
3606 """ The Network Service Manager class"""
3607 def __init__(self
, dts
, log
, loop
,
3608 nsr_handler
, vnfr_handler
, vlr_handler
, ro_plugin_selector
,
3609 vnffgmgr
, vnfd_pub_handler
, cloud_account_handler
):
3613 self
._nsr
_handler
= nsr_handler
3614 self
._vnfr
_pub
_handler
= vnfr_handler
3615 self
._vlr
_pub
_handler
= vlr_handler
3616 self
._vnffgmgr
= vnffgmgr
3617 self
._vnfd
_pub
_handler
= vnfd_pub_handler
3618 self
._cloud
_account
_handler
= cloud_account_handler
3620 self
._ro
_plugin
_selector
= ro_plugin_selector
3621 self
._ncclient
= rift
.mano
.ncclient
.NcClient(
3633 self
.cfgmgr_obj
= conman
.ROConfigManager(log
, loop
, dts
, self
)
3635 # TODO: All these handlers should move to tasklet level.
3636 # Passing self is often an indication of bad design
3637 self
._nsd
_dts
_handler
= NsdDtsHandler(dts
, log
, loop
, self
)
3638 self
._vnfd
_dts
_handler
= VnfdDtsHandler(dts
, log
, loop
, self
)
3639 self
._dts
_handlers
= [self
._nsd
_dts
_handler
,
3640 VnfrDtsHandler(dts
, log
, loop
, self
),
3641 NsdRefCountDtsHandler(dts
, log
, loop
, self
),
3642 NsrDtsHandler(dts
, log
, loop
, self
),
3643 ScalingRpcHandler(log
, dts
, loop
, self
.scale_rpc_callback
),
3644 NsrRpcDtsHandler(dts
,log
,loop
,self
),
3645 self
._vnfd
_dts
_handler
,
3666 def nsr_handler(self
):
3667 """" NSR handler """
3668 return self
._nsr
_handler
3672 """" So Obj handler """
3677 """ NSRs in this NSM"""
3682 """ NSDs in this NSM"""
3687 """ VNFDs in this NSM"""
3692 """ VNFRs in this NSM"""
3696 def nsr_pub_handler(self
):
3697 """ NSR publication handler """
3698 return self
._nsr
_handler
3701 def vnfr_pub_handler(self
):
3702 """ VNFR publication handler """
3703 return self
._vnfr
_pub
_handler
3706 def vlr_pub_handler(self
):
3707 """ VLR publication handler """
3708 return self
._vlr
_pub
_handler
3711 def vnfd_pub_handler(self
):
3712 return self
._vnfd
_pub
_handler
3716 """ Register all static DTS handlers """
3717 for dts_handle
in self
._dts
_handlers
:
3718 yield from dts_handle
.register()
3721 def get_ns_by_nsr_id(self
, nsr_id
):
3722 """ get NSR by nsr id """
3723 if nsr_id
not in self
._nsrs
:
3724 raise NetworkServiceRecordError("NSR id %s not found" % nsr_id
)
3726 return self
._nsrs
[nsr_id
]
3728 def scale_nsr_out(self
, nsr_id
, scale_group_name
, instance_id
, config_xact
):
3729 self
.log
.debug("Scale out NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3734 nsr
= self
._nsrs
[nsr_id
]
3735 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3736 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3738 self
._loop
.create_task(nsr
.create_scale_group_instance(scale_group_name
, instance_id
, config_xact
))
3740 def scale_nsr_in(self
, nsr_id
, scale_group_name
, instance_id
):
3741 self
.log
.debug("Scale in NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3746 nsr
= self
._nsrs
[nsr_id
]
3747 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3748 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3750 self
._loop
.create_task(nsr
.delete_scale_group_instance(scale_group_name
, instance_id
))
3752 def scale_rpc_callback(self
, xact
, msg
, action
):
3753 """Callback handler for RPC calls
3755 xact : Transaction Handler
3757 action : Scaling Action
3759 ScalingGroupInstance
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup_Instance
3760 ScalingGroup
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
3762 xpath
= ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').format(
3764 instance
= ScalingGroupInstance
.from_dict({"id": msg
.instance_id
})
3767 def get_nsr_scaling_group():
3768 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3770 for result
in results
:
3771 res
= yield from result
3772 nsr_config
= res
.result
3774 for scaling_group
in nsr_config
.scaling_group
:
3775 if scaling_group
.scaling_group_name_ref
== msg
.scaling_group_name_ref
:
3778 scaling_group
= nsr_config
.scaling_group
.add()
3779 scaling_group
.scaling_group_name_ref
= msg
.scaling_group_name_ref
3781 return (nsr_config
, scaling_group
)
3784 def update_config(nsr_config
):
3785 xml
= self
._ncclient
.convert_to_xml(RwNsrYang
, nsr_config
)
3786 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
3787 yield from self
._ncclient
.connect()
3788 yield from self
._ncclient
.manager
.edit_config(target
="running", config
=xml
, default_operation
="replace")
3792 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3793 scaling_group
.instance
.append(instance
)
3794 yield from update_config(nsr_config
)
3798 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3799 scaling_group
.instance
.remove(instance
)
3800 yield from update_config(nsr_config
)
3802 if action
== ScalingRpcHandler
.ACTION
.SCALE_OUT
:
3803 self
._loop
.create_task(scale_out())
3805 self
._loop
.create_task(scale_in())
3807 # Opdata based calls, disabled for now!
3808 # if action == ScalingRpcHandler.ACTION.SCALE_OUT:
3809 # self.scale_nsr_out(
3811 # msg.scaling_group_name_ref,
3815 # self.scale_nsr_in(
3817 # msg.scaling_group_name_ref,
3820 def nsr_update_cfg(self
, nsr_id
, msg
):
3821 nsr
= self
._nsrs
[nsr_id
]
3822 nsr
.nsr_cfg_msg
= msg
3824 def nsr_instantiate_vl(self
, nsr_id
, vld
):
3825 self
.log
.debug("NSR {} create VL {}".format(nsr_id
, vld
))
3826 nsr
= self
._nsrs
[nsr_id
]
3827 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3828 raise NsrVlUpdateError("Cannot perform VL instantiate if NSR is not in running state")
3830 # Not calling in a separate task as this is called from a separate task
3831 yield from nsr
.create_vl_instance(vld
)
3833 def nsr_terminate_vl(self
, nsr_id
, vld
):
3834 self
.log
.debug("NSR {} delete VL {}".format(nsr_id
, vld
.id))
3835 nsr
= self
._nsrs
[nsr_id
]
3836 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3837 raise NsrVlUpdateError("Cannot perform VL terminate if NSR is not in running state")
3839 # Not calling in a separate task as this is called from a separate task
3840 yield from nsr
.delete_vl_instance(vld
)
3842 def create_nsr(self
, nsr_msg
, key_pairs
=None,restart_mode
=False):
3843 """ Create an NSR instance """
3844 if nsr_msg
.id in self
._nsrs
:
3845 msg
= "NSR id %s already exists" % nsr_msg
.id
3846 self
._log
.error(msg
)
3847 raise NetworkServiceRecordError(msg
)
3849 self
._log
.info("Create NetworkServiceRecord nsr id %s from nsd_id %s",
3853 nsm_plugin
= self
._ro
_plugin
_selector
.ro_plugin
3854 sdn_account_name
= self
._cloud
_account
_handler
.get_cloud_account_sdn_name(nsr_msg
.cloud_account
)
3856 nsr
= NetworkServiceRecord(self
._dts
,
3864 restart_mode
=restart_mode
,
3865 vlr_handler
=self
._ro
_plugin
_selector
._records
_publisher
._vlr
_pub
_hdlr
3867 self
._nsrs
[nsr_msg
.id] = nsr
3868 nsm_plugin
.create_nsr(nsr_msg
, nsr_msg
.nsd
, key_pairs
)
3872 def delete_nsr(self
, nsr_id
):
3874 Delete NSR with the passed nsr id
3876 del self
._nsrs
[nsr_id
]
3879 def instantiate_ns(self
, nsr_id
, config_xact
):
3880 """ Instantiate an NS instance """
3881 self
._log
.debug("Instantiating Network service id %s", nsr_id
)
3882 if nsr_id
not in self
._nsrs
:
3883 err
= "NSR id %s not found " % nsr_id
3884 self
._log
.error(err
)
3885 raise NetworkServiceRecordError(err
)
3887 nsr
= self
._nsrs
[nsr_id
]
3888 yield from nsr
.nsm_plugin
.instantiate_ns(nsr
, config_xact
)
3891 def update_vnfr(self
, vnfr
):
3892 """Create/Update an VNFR """
3894 vnfr_state
= self
._vnfrs
[vnfr
.id].state
3895 self
._log
.debug("Updating VNFR with state %s: vnfr %s", vnfr_state
, vnfr
)
3897 yield from self
._vnfrs
[vnfr
.id].update_state(vnfr
)
3898 nsr
= self
.find_nsr_for_vnfr(vnfr
.id)
3899 yield from nsr
.update_state()
3901 def find_nsr_for_vnfr(self
, vnfr_id
):
3902 """ Find the NSR which )has the passed vnfr id"""
3903 for nsr
in list(self
.nsrs
.values()):
3904 for vnfr
in list(nsr
.vnfrs
.values()):
3905 if vnfr
.id == vnfr_id
:
3909 def delete_vnfr(self
, vnfr_id
):
3910 """ Delete VNFR with the passed id"""
3911 del self
._vnfrs
[vnfr_id
]
3913 def get_nsd_ref(self
, nsd_id
):
3914 """ Get network service descriptor for the passed nsd_id
3916 nsd
= self
.get_nsd(nsd_id
)
3921 def get_nsr_config(self
, nsd_id
):
3922 xpath
= "C,/nsr:ns-instance-config"
3923 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3925 for result
in results
:
3926 entry
= yield from result
3927 ns_instance_config
= entry
.result
3929 for nsr
in ns_instance_config
.nsr
:
3930 if nsr
.nsd
.id == nsd_id
:
3936 def nsd_unref_by_nsr_id(self
, nsr_id
):
3937 """ Unref the network service descriptor based on NSR id """
3938 self
._log
.debug("NSR Unref called for Nsr Id:%s", nsr_id
)
3939 if nsr_id
in self
._nsrs
:
3940 nsr
= self
._nsrs
[nsr_id
]
3943 nsd
= self
.get_nsd(nsr
.nsd_id
)
3944 self
._log
.debug("Releasing ref on NSD %s held by NSR %s - Curr %d",
3945 nsd
.id, nsr
.id, nsd
.ref_count
)
3947 except NetworkServiceDescriptorError
:
3948 # We store a copy of NSD in NSR and the NSD in nsd-catalog
3953 self
._log
.error("Cannot find NSR with id %s", nsr_id
)
3954 raise NetworkServiceDescriptorUnrefError("No NSR with id" % nsr_id
)
3957 def nsd_unref(self
, nsd_id
):
3958 """ Unref the network service descriptor associated with the id """
3959 nsd
= self
.get_nsd(nsd_id
)
3962 def get_nsd(self
, nsd_id
):
3963 """ Get network service descriptor for the passed nsd_id"""
3964 if nsd_id
not in self
._nsds
:
3965 self
._log
.error("Cannot find NSD id:%s", nsd_id
)
3966 raise NetworkServiceDescriptorError("Cannot find NSD id:%s", nsd_id
)
3968 return self
._nsds
[nsd_id
]
3970 def create_nsd(self
, nsd_msg
):
3971 """ Create a network service descriptor """
3972 self
._log
.debug("Create network service descriptor - %s", nsd_msg
)
3973 if nsd_msg
.id in self
._nsds
:
3974 self
._log
.error("Cannot create NSD %s -NSD ID already exists", nsd_msg
)
3975 raise NetworkServiceDescriptorError("NSD already exists-%s", nsd_msg
.id)
3977 nsd
= NetworkServiceDescriptor(
3984 self
._nsds
[nsd_msg
.id] = nsd
3988 def update_nsd(self
, nsd
):
3989 """ update the Network service descriptor """
3990 self
._log
.debug("Update network service descriptor - %s", nsd
)
3991 if nsd
.id not in self
._nsds
:
3992 self
._log
.debug("No NSD found - creating NSD id = %s", nsd
.id)
3993 self
.create_nsd(nsd
)
3995 self
._log
.debug("Updating NSD id = %s, nsd = %s", nsd
.id, nsd
)
3996 self
._nsds
[nsd
.id].update(nsd
)
3998 def delete_nsd(self
, nsd_id
):
3999 """ Delete the Network service descriptor with the passed id """
4000 self
._log
.debug("Deleting the network service descriptor - %s", nsd_id
)
4001 if nsd_id
not in self
._nsds
:
4002 self
._log
.debug("Delete NSD failed - cannot find nsd-id %s", nsd_id
)
4003 raise NetworkServiceDescriptorNotFound("Cannot find %s", nsd_id
)
4005 if nsd_id
not in self
._nsds
:
4006 self
._log
.debug("Cannot delete NSD id %s reference exists %s",
4008 self
._nsds
[nsd_id
].ref_count
)
4009 raise NetworkServiceDescriptorRefCountExists(
4010 "Cannot delete :%s, ref_count:%s",
4012 self
._nsds
[nsd_id
].ref_count
)
4014 del self
._nsds
[nsd_id
]
4016 def get_vnfd_config(self
, xact
):
4017 vnfd_dts_reg
= self
._vnfd
_dts
_handler
.regh
4018 for cfg
in vnfd_dts_reg
.get_xact_elements(xact
):
4019 if cfg
.id not in self
._vnfds
:
4020 self
.create_vnfd(cfg
)
4022 def get_vnfd(self
, vnfd_id
, xact
):
4023 """ Get virtual network function descriptor for the passed vnfd_id"""
4024 if vnfd_id
not in self
._vnfds
:
4025 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4026 self
.get_vnfd_config(xact
)
4028 if vnfd_id
not in self
._vnfds
:
4029 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4030 raise VnfDescriptorError("Cannot find VNFD id:%s", vnfd_id
)
4032 return self
._vnfds
[vnfd_id
]
4034 def create_vnfd(self
, vnfd
):
4035 """ Create a virtual network function descriptor """
4036 self
._log
.debug("Create virtual network function descriptor - %s", vnfd
)
4037 if vnfd
.id in self
._vnfds
:
4038 self
._log
.error("Cannot create VNFD %s -VNFD ID already exists", vnfd
)
4039 raise VnfDescriptorError("VNFD already exists-%s", vnfd
.id)
4041 self
._vnfds
[vnfd
.id] = vnfd
4042 return self
._vnfds
[vnfd
.id]
4044 def update_vnfd(self
, vnfd
):
4045 """ Update the virtual network function descriptor """
4046 self
._log
.debug("Update virtual network function descriptor- %s", vnfd
)
4049 if vnfd
.id not in self
._vnfds
:
4050 self
._log
.debug("No VNFD found - creating VNFD id = %s", vnfd
.id)
4051 self
.create_vnfd(vnfd
)
4053 self
._log
.debug("Updating VNFD id = %s, vnfd = %s", vnfd
.id, vnfd
)
4054 self
._vnfds
[vnfd
.id] = vnfd
4057 def delete_vnfd(self
, vnfd_id
):
4058 """ Delete the virtual network function descriptor with the passed id """
4059 self
._log
.debug("Deleting the virtual network function descriptor - %s", vnfd_id
)
4060 if vnfd_id
not in self
._vnfds
:
4061 self
._log
.debug("Delete VNFD failed - cannot find vnfd-id %s", vnfd_id
)
4062 raise VnfDescriptorError("Cannot find %s", vnfd_id
)
4064 del self
._vnfds
[vnfd_id
]
4066 def nsd_in_use(self
, nsd_id
):
4067 """ Is the NSD with the passed id in use """
4068 self
._log
.debug("Is this NSD in use - msg:%s", nsd_id
)
4069 if nsd_id
in self
._nsds
:
4070 return self
._nsds
[nsd_id
].in_use()
4074 def publish_nsr(self
, xact
, path
, msg
):
4075 """ Publish a NSR """
4076 self
._log
.debug("Publish NSR with path %s, msg %s",
4078 yield from self
.nsr_handler
.update(xact
, path
, msg
)
4081 def unpublish_nsr(self
, xact
, path
):
4082 """ Un Publish an NSR """
4083 self
._log
.debug("Publishing delete NSR with path %s", path
)
4084 yield from self
.nsr_handler
.delete(path
, xact
)
4086 def vnfr_is_ready(self
, vnfr_id
):
4087 """ VNFR with the id is ready """
4088 self
._log
.debug("VNFR id %s ready", vnfr_id
)
4089 if vnfr_id
not in self
._vnfds
:
4090 err
= "Did not find VNFR ID with id %s" % vnfr_id
4091 self
._log
.critical("err")
4092 raise VirtualNetworkFunctionRecordError(err
)
4093 self
._vnfrs
[vnfr_id
].is_ready()
4096 def get_nsd_refcount(self
, nsd_id
):
4097 """ Get the nsd_list from this NSM"""
4099 def nsd_refcount_xpath(nsd_id
):
4100 """ xpath for ref count entry """
4101 return (NsdRefCountDtsHandler
.XPATH
+
4102 "[rw-nsr:nsd-id-ref = '{}']").format(nsd_id
)
4105 if nsd_id
is None or nsd_id
== "":
4106 for nsd
in self
._nsds
.values():
4107 nsd_msg
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4108 nsd_msg
.nsd_id_ref
= nsd
.id
4109 nsd_msg
.instance_ref_count
= nsd
.ref_count
4110 nsd_list
.append((nsd_refcount_xpath(nsd
.id), nsd_msg
))
4111 elif nsd_id
in self
._nsds
:
4112 nsd_msg
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4113 nsd_msg
.nsd_id_ref
= self
._nsds
[nsd_id
].id
4114 nsd_msg
.instance_ref_count
= self
._nsds
[nsd_id
].ref_count
4115 nsd_list
.append((nsd_refcount_xpath(nsd_id
), nsd_msg
))
4120 def terminate_ns(self
, nsr_id
, xact
):
4122 Terminate network service for the given NSR Id
4125 # Terminate the instances/networks assocaited with this nw service
4126 self
._log
.debug("Terminating the network service %s", nsr_id
)
4127 yield from self
._nsrs
[nsr_id
].terminate()
4130 yield from self
.nsd_unref_by_nsr_id(nsr_id
)
4132 # Unpublish the NSR record
4133 self
._log
.debug("Unpublishing the network service %s", nsr_id
)
4134 yield from self
._nsrs
[nsr_id
].unpublish(xact
)
4136 # Finaly delete the NS instance from this NS Manager
4137 self
._log
.debug("Deletng the network service %s", nsr_id
)
4138 self
.delete_nsr(nsr_id
)
4141 class NsmRecordsPublisherProxy(object):
4142 """ This class provides a publisher interface that allows plugin objects
4143 to publish NSR/VNFR/VLR"""
4145 def __init__(self
, dts
, log
, loop
, nsr_pub_hdlr
, vnfr_pub_hdlr
, vlr_pub_hdlr
):
4149 self
._nsr
_pub
_hdlr
= nsr_pub_hdlr
4150 self
._vlr
_pub
_hdlr
= vlr_pub_hdlr
4151 self
._vnfr
_pub
_hdlr
= vnfr_pub_hdlr
4154 def publish_nsr(self
, xact
, nsr
):
4155 """ Publish an NSR """
4156 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4157 return (yield from self
._nsr
_pub
_hdlr
.update(xact
, path
, nsr
))
4160 def unpublish_nsr(self
, xact
, nsr
):
4161 """ Unpublish an NSR """
4162 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4163 return (yield from self
._nsr
_pub
_hdlr
.delete(xact
, path
))
4166 def publish_vnfr(self
, xact
, vnfr
):
4167 """ Publish an VNFR """
4168 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4169 return (yield from self
._vnfr
_pub
_hdlr
.update(xact
, path
, vnfr
))
4172 def unpublish_vnfr(self
, xact
, vnfr
):
4173 """ Unpublish a VNFR """
4174 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4175 return (yield from self
._vnfr
_pub
_hdlr
.delete(xact
, path
))
4178 def publish_vlr(self
, xact
, vlr
):
4179 """ Publish a VLR """
4180 path
= VirtualLinkRecord
.vlr_xpath(vlr
)
4181 return (yield from self
._vlr
_pub
_hdlr
.update(xact
, path
, vlr
))
4184 def unpublish_vlr(self
, xact
, vlr
):
4185 """ Unpublish a VLR """
4186 path
= VirtualLinkRecord
.vlr_xpath(vlr
)
4187 return (yield from self
._vlr
_pub
_hdlr
.delete(xact
, path
))
4190 class ScalingRpcHandler(mano_dts
.DtsHandler
):
4191 """ The Network service Monitor DTS handler """
4192 SCALE_IN_INPUT_XPATH
= "I,/nsr:exec-scale-in"
4193 SCALE_IN_OUTPUT_XPATH
= "O,/nsr:exec-scale-in"
4195 SCALE_OUT_INPUT_XPATH
= "I,/nsr:exec-scale-out"
4196 SCALE_OUT_OUTPUT_XPATH
= "O,/nsr:exec-scale-out"
4198 ACTION
= Enum('ACTION', 'SCALE_IN SCALE_OUT')
4200 def __init__(self
, log
, dts
, loop
, callback
=None):
4201 super().__init
__(log
, dts
, loop
)
4202 self
.callback
= callback
4203 self
.last_instance_id
= defaultdict(int)
4209 def on_scale_in_prepare(xact_info
, action
, ks_path
, msg
):
4210 assert action
== rwdts
.QueryAction
.RPC
4214 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_IN
)
4216 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleIn
.from_dict({
4217 "instance_id": msg
.instance_id
})
4219 xact_info
.respond_xpath(
4220 rwdts
.XactRspCode
.ACK
,
4221 self
.__class
__.SCALE_IN_OUTPUT_XPATH
,
4224 except Exception as e
:
4225 self
.log
.exception(e
)
4226 xact_info
.respond_xpath(
4227 rwdts
.XactRspCode
.NACK
,
4228 self
.__class
__.SCALE_IN_OUTPUT_XPATH
)
4231 def on_scale_out_prepare(xact_info
, action
, ks_path
, msg
):
4232 assert action
== rwdts
.QueryAction
.RPC
4235 scaling_group
= msg
.scaling_group_name_ref
4236 if not msg
.instance_id
:
4237 last_instance_id
= self
.last_instance_id
[scale_group
]
4238 msg
.instance_id
= last_instance_id
+ 1
4239 self
.last_instance_id
[scale_group
] += 1
4242 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_OUT
)
4244 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleOut
.from_dict({
4245 "instance_id": msg
.instance_id
})
4247 xact_info
.respond_xpath(
4248 rwdts
.XactRspCode
.ACK
,
4249 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
,
4252 except Exception as e
:
4253 self
.log
.exception(e
)
4254 xact_info
.respond_xpath(
4255 rwdts
.XactRspCode
.NACK
,
4256 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
)
4258 scale_in_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4259 on_prepare
=on_scale_in_prepare
)
4260 scale_out_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4261 on_prepare
=on_scale_out_prepare
)
4263 with self
.dts
.group_create() as group
:
4265 xpath
=self
.__class
__.SCALE_IN_INPUT_XPATH
,
4266 handler
=scale_in_hdl
,
4267 flags
=rwdts
.Flag
.PUBLISHER
)
4269 xpath
=self
.__class
__.SCALE_OUT_INPUT_XPATH
,
4270 handler
=scale_out_hdl
,
4271 flags
=rwdts
.Flag
.PUBLISHER
)
4274 class NsmTasklet(rift
.tasklets
.Tasklet
):
4276 The network service manager tasklet
4278 def __init__(self
, *args
, **kwargs
):
4279 super(NsmTasklet
, self
).__init
__(*args
, **kwargs
)
4280 self
.rwlog
.set_category("rw-mano-log")
4281 self
.rwlog
.set_subcategory("nsm")
4286 self
._ro
_plugin
_selector
= None
4287 self
._vnffgmgr
= None
4289 self
._nsr
_handler
= None
4290 self
._vnfr
_pub
_handler
= None
4291 self
._vlr
_pub
_handler
= None
4292 self
._vnfd
_pub
_handler
= None
4293 self
._scale
_cfg
_handler
= None
4295 self
._records
_publisher
_proxy
= None
4298 """ The task start callback """
4299 super(NsmTasklet
, self
).start()
4300 self
.log
.info("Starting NsmTasklet")
4302 self
.log
.debug("Registering with dts")
4303 self
._dts
= rift
.tasklets
.DTS(self
.tasklet_info
,
4304 RwNsmYang
.get_schema(),
4306 self
.on_dts_state_change
)
4308 self
.log
.debug("Created DTS Api GI Object: %s", self
._dts
)
4314 print("Caught Exception in NSM stop:", sys
.exc_info()[0])
4317 def on_instance_started(self
):
4318 """ Task instance started callback """
4319 self
.log
.debug("Got instance started callback")
4323 """ Task init callback """
4324 self
.log
.debug("Got instance started callback")
4326 self
.log
.debug("creating config account handler")
4328 self
._nsr
_pub
_handler
= publisher
.NsrOpDataDtsHandler(self
._dts
, self
.log
, self
.loop
)
4329 yield from self
._nsr
_pub
_handler
.register()
4331 self
._vnfr
_pub
_handler
= publisher
.VnfrPublisherDtsHandler(self
._dts
, self
.log
, self
.loop
)
4332 yield from self
._vnfr
_pub
_handler
.register()
4334 self
._vlr
_pub
_handler
= publisher
.VlrPublisherDtsHandler(self
._dts
, self
.log
, self
.loop
)
4335 yield from self
._vlr
_pub
_handler
.register()
4337 manifest
= self
.tasklet_info
.get_pb_manifest()
4338 use_ssl
= manifest
.bootstrap_phase
.rwsecurity
.use_ssl
4339 ssl_cert
= manifest
.bootstrap_phase
.rwsecurity
.cert
4340 ssl_key
= manifest
.bootstrap_phase
.rwsecurity
.key
4342 self
._vnfd
_pub
_handler
= publisher
.VnfdPublisher(use_ssl
, ssl_cert
, ssl_key
, self
.loop
)
4344 self
._records
_publisher
_proxy
= NsmRecordsPublisherProxy(
4348 self
._nsr
_pub
_handler
,
4349 self
._vnfr
_pub
_handler
,
4350 self
._vlr
_pub
_handler
,
4353 # Register the NSM to receive the nsm plugin
4354 # when cloud account is configured
4355 self
._ro
_plugin
_selector
= cloud
.ROAccountPluginSelector(
4359 self
._records
_publisher
_proxy
,
4361 yield from self
._ro
_plugin
_selector
.register()
4363 self
._cloud
_account
_handler
= cloud
.CloudAccountConfigSubscriber(
4368 yield from self
._cloud
_account
_handler
.register()
4370 self
._vnffgmgr
= rwvnffgmgr
.VnffgMgr(self
._dts
, self
.log
, self
.log_hdl
, self
.loop
)
4371 yield from self
._vnffgmgr
.register()
4373 self
._nsm
= NsManager(
4377 self
._nsr
_pub
_handler
,
4378 self
._vnfr
_pub
_handler
,
4379 self
._vlr
_pub
_handler
,
4380 self
._ro
_plugin
_selector
,
4382 self
._vnfd
_pub
_handler
,
4383 self
._cloud
_account
_handler
4386 yield from self
._nsm
.register()
4390 """ Task run callback """
4394 def on_dts_state_change(self
, state
):
4395 """Take action according to current dts state to transition
4396 application into the corresponding application state
4399 state - current dts state
4402 rwdts
.State
.INIT
: rwdts
.State
.REGN_COMPLETE
,
4403 rwdts
.State
.CONFIG
: rwdts
.State
.RUN
,
4407 rwdts
.State
.INIT
: self
.init
,
4408 rwdts
.State
.RUN
: self
.run
,
4411 # Transition application to next state
4412 handler
= handlers
.get(state
, None)
4413 if handler
is not None:
4414 yield from handler()
4416 # Transition dts to next state
4417 next_state
= switch
.get(state
, None)
4418 if next_state
is not None:
4419 self
.log
.debug("Changing state to %s", next_state
)
4420 self
._dts
.handle
.set_state(next_state
)