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
800 self
._prev
_state
= VnfRecordState
.INIT
801 self
._state
= VnfRecordState
.INIT
802 self
._state
_failed
_reason
= None
804 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
807 self
._vnfr
_id
= str(uuid
.uuid4())
809 self
._vnfr
_msg
= self
.create_vnfr_msg()
810 self
._log
.debug("Set VNFR {} config type to {}".
811 format(self
.name
, self
.config_type
))
812 self
.restart_mode
= restart_mode
815 if group_name
is None and group_instance_id
is not None:
816 raise ValueError("Group instance id must not be provided with an empty group name")
826 return "D,/vnfr:vnfr-catalog/vnfr:vnfr[vnfr:id = '{}']".format(self
.id)
831 return self
._vnfr
_msg
834 def const_vnfr_msg(self
):
836 return RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_ConstituentVnfrRef(vnfr_id
=self
.id,cloud_account
=self
.cloud_account_name
,om_datacenter
=self
._om
_datacenter
_name
)
844 def cloud_account_name(self
):
845 """ Cloud account that this VNF should be created in """
846 return self
._cloud
_account
_name
849 def om_datacenter_name(self
):
850 """ Datacenter that this VNF should be created in """
851 return self
._om
_datacenter
_name
856 """ Is this VNF actve """
857 return True if self
._state
== VnfRecordState
.ACTIVE
else False
861 """ state of this VNF """
865 def state_failed_reason(self
):
866 """ Error message in case this VNF is in failed state """
867 return self
._state
_failed
_reason
870 def member_vnf_index(self
):
871 """ Member VNF index """
872 return self
._const
_vnfd
_msg
.member_vnf_index
877 return self
._nsr
_name
881 """ Name of this VNFR """
882 if self
._name
is not None:
885 name_tags
= [self
._nsr
_name
]
887 if self
._group
_name
is not None:
888 name_tags
.append(self
._group
_name
)
890 if self
._group
_instance
_id
is not None:
891 name_tags
.append(str(self
._group
_instance
_id
))
893 name_tags
.extend([self
.vnfd
.name
, str(self
.member_vnf_index
)])
895 self
._name
= "__".join(name_tags
)
900 def vnfr_xpath(vnfr
):
901 """ Get the VNFR path from VNFR """
902 return (VirtualNetworkFunctionRecord
.XPATH
+ "[vnfr:id = '{}']").format(vnfr
.id)
905 def config_type(self
):
906 cfg_types
= ['netconf', 'juju', 'script']
907 for method
in cfg_types
:
908 if self
._vnfd
.vnf_configuration
.has_field(method
):
913 def config_status(self
):
914 """Return the config status as YANG ENUM string"""
915 self
._log
.debug("Map VNFR {} config status {} ({})".
916 format(self
.name
, self
._config
_status
, self
.config_type
))
917 if self
.config_type
== 'none':
918 return 'config_not_needed'
919 elif self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
921 elif self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
926 def set_state(self
, state
):
927 """ set the state of this object """
928 self
._prev
_state
= self
._state
931 def reset_id(self
, vnfr_id
):
932 self
._vnfr
_id
= vnfr_id
933 self
._vnfr
_msg
= self
.create_vnfr_msg()
936 self
.config_store
.merge_vnfd_config(
939 self
.member_vnf_index
,
942 def create_vnfr_msg(self
):
943 """ VNFR message for this VNFR """
951 vnfd_copy_dict
= {k
: v
for k
, v
in self
._vnfd
.as_dict().items() if k
in vnfd_fields
}
954 "nsr_id_ref": self
._nsr
_id
,
955 "vnfd_ref": self
.vnfd
.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
)
964 vnfr
.member_vnf_index_ref
= self
.member_vnf_index
965 vnfr
.vnf_configuration
.from_dict(self
._vnfd
.vnf_configuration
.as_dict())
967 if self
._vnfd
.mgmt_interface
.has_field("port"):
968 vnfr
.mgmt_interface
.port
= self
._vnfd
.mgmt_interface
.port
970 for group_info
in self
._placement
_groups
:
971 group
= vnfr
.placement_groups_info
.add()
972 group
.from_dict(group_info
.as_dict())
974 # UI expects the monitoring param field to exist
975 vnfr
.monitoring_param
= []
977 self
._log
.debug("Get vnfr_msg for VNFR {} : {}".format(self
.name
, vnfr
))
981 def update_vnfm(self
):
982 self
._log
.debug("Send an update to VNFM for VNFR {} with {}".
983 format(self
.name
, self
.vnfr_msg
))
984 yield from self
._dts
.query_update(
986 rwdts
.XactFlag
.TRACE
,
990 def get_config_status(self
):
991 """Return the config status as YANG ENUM"""
992 return self
._config
_status
995 def set_config_status(self
, status
):
997 def status_to_string(status
):
999 NsrYang
.ConfigStates
.INIT
: 'init',
1000 NsrYang
.ConfigStates
.CONFIGURING
: 'configuring',
1001 NsrYang
.ConfigStates
.CONFIG_NOT_NEEDED
: 'config_not_needed',
1002 NsrYang
.ConfigStates
.CONFIGURED
: 'configured',
1003 NsrYang
.ConfigStates
.FAILED
: 'failed',
1006 return status_dc
[status
]
1008 self
._log
.debug("Update VNFR {} from {} ({}) to {}".
1009 format(self
.name
, self
._config
_status
,
1010 self
.config_type
, status
))
1011 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1012 self
._log
.error("Updating already configured VNFR {}".
1016 if self
._config
_status
!= status
:
1018 self
._config
_status
= status
1019 # I don't think this is used. Original implementor can check.
1020 # Caused Exception, so corrected it by status_to_string
1021 # But not sure whats the use of this variable?
1022 self
.vnfr_msg
.config_status
= status_to_string(status
)
1023 except Exception as e
:
1024 self
._log
.error("Exception=%s", str(e
))
1027 self
._log
.debug("Updated VNFR {} status to {}".format(self
.name
, status
))
1029 if self
._config
_status
!= NsrYang
.ConfigStates
.INIT
:
1031 # Publish only after VNFM has the VNFR created
1032 yield from self
.update_vnfm()
1033 except Exception as e
:
1034 self
._log
.error("Exception updating VNFM with new status {} of VNFR {}: {}".
1035 format(status
, self
.name
, e
))
1036 self
._log
.exception(e
)
1038 def is_configured(self
):
1039 if self
.config_type
== 'none':
1042 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1048 def instantiate(self
, nsr
):
1049 """ Instantiate this VNFR"""
1051 self
._log
.debug("Instaniating VNFR key %s, vnfd %s",
1052 self
.xpath
, self
._vnfd
)
1054 self
._log
.debug("Create VNF with xpath %s and vnfr %s",
1055 self
.xpath
, self
.vnfr_msg
)
1057 self
.set_state(VnfRecordState
.INSTANTIATION_PENDING
)
1059 def find_vlr_for_cp(conn
):
1060 """ Find VLR for the given connection point """
1061 for vlr
in nsr
.vlrs
:
1062 for vnfd_cp
in vlr
.vld_msg
.vnfd_connection_point_ref
:
1063 if (vnfd_cp
.vnfd_id_ref
== self
._vnfd
.id and
1064 vnfd_cp
.vnfd_connection_point_ref
== conn
.name
and
1065 vnfd_cp
.member_vnf_index_ref
== self
.member_vnf_index
and
1066 vlr
.cloud_account_name
== self
.cloud_account_name
):
1067 self
._log
.debug("Found VLR for cp_name:%s and vnf-index:%d",
1068 conn
.name
, self
.member_vnf_index
)
1072 # For every connection point in the VNFD fill in the identifier
1073 for conn_p
in self
._vnfd
.connection_point
:
1074 cpr
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_ConnectionPoint()
1075 cpr
.name
= conn_p
.name
1076 cpr
.type_yang
= conn_p
.type_yang
1077 vlr_ref
= find_vlr_for_cp(conn_p
)
1079 msg
= "Failed to find VLR for cp = %s" % conn_p
.name
1080 self
._log
.debug("%s", msg
)
1081 # raise VirtualNetworkFunctionRecordError(msg)
1084 cpr
.vlr_ref
= vlr_ref
.id
1085 self
.vnfr_msg
.connection_point
.append(cpr
)
1086 self
._log
.debug("Connection point [%s] added, vnf id=%s vnfd id=%s",
1087 cpr
, self
.vnfr_msg
.id, self
.vnfr_msg
.vnfd_ref
)
1089 if not self
.restart_mode
:
1090 yield from self
._dts
.query_create(self
.xpath
,
1094 yield from self
._dts
.query_update(self
.xpath
,
1098 self
._log
.info("Created VNF with xpath %s and vnfr %s",
1099 self
.xpath
, self
.vnfr_msg
)
1101 self
._log
.info("Instantiated VNFR with xpath %s and vnfd %s, vnfr %s",
1102 self
.xpath
, self
._vnfd
, self
.vnfr_msg
)
1105 def update_state(self
, vnfr_msg
):
1106 """ Update this VNFR"""
1107 if vnfr_msg
.operational_status
== "running":
1108 if self
.vnfr_msg
.operational_status
!= "running":
1109 yield from self
.is_active()
1110 elif vnfr_msg
.operational_status
== "failed":
1111 yield from self
.instantiation_failed(failed_reason
=vnfr_msg
.operational_status_details
)
1114 def is_active(self
):
1115 """ This VNFR is active """
1116 self
._log
.debug("VNFR %s is active", self
._vnfr
_id
)
1117 self
.set_state(VnfRecordState
.ACTIVE
)
1120 def instantiation_failed(self
, failed_reason
=None):
1121 """ This VNFR instantiation failed"""
1122 self
._log
.error("VNFR %s instantiation failed", self
._vnfr
_id
)
1123 self
.set_state(VnfRecordState
.FAILED
)
1124 self
._state
_failed
_reason
= failed_reason
1126 def vnfr_in_vnfm(self
):
1127 """ Is there a VNFR record in VNFM """
1128 if (self
._state
== VnfRecordState
.ACTIVE
or
1129 self
._state
== VnfRecordState
.INSTANTIATION_PENDING
or
1130 self
._state
== VnfRecordState
.FAILED
):
1136 def terminate(self
):
1137 """ Terminate this VNF """
1138 if not self
.vnfr_in_vnfm():
1139 self
._log
.debug("Ignoring terminate request for id %s in state %s",
1140 self
.id, self
._state
)
1143 self
._log
.debug("Terminating VNF id:%s", self
.id)
1144 self
.set_state(VnfRecordState
.TERMINATE_PENDING
)
1145 with self
._dts
.transaction(flags
=0) as xact
:
1146 block
= xact
.block_create()
1147 block
.add_query_delete(self
.xpath
)
1148 yield from block
.execute(flags
=0)
1149 self
.set_state(VnfRecordState
.TERMINATED
)
1150 self
._log
.debug("Terminated VNF id:%s", self
.id)
1153 class NetworkServiceStatus(object):
1154 """ A class representing the Network service's status """
1155 MAX_EVENTS_RECORDED
= 10
1156 """ Network service Status class"""
1157 def __init__(self
, dts
, log
, loop
):
1162 self
._state
= NetworkServiceRecordState
.INIT
1163 self
._events
= deque([])
1166 def create_notification(self
, evt
, evt_desc
, evt_details
):
1167 xp
= "N,/rw-nsr:nsm-notification"
1168 notif
= RwNsrYang
.YangNotif_RwNsr_NsmNotification()
1170 notif
.description
= evt_desc
1171 notif
.details
= evt_details
if evt_details
is not None else None
1173 yield from self
._dts
.query_create(xp
, rwdts
.XactFlag
.ADVISE
, notif
)
1174 self
._log
.info("Notification called by creating dts query: %s", notif
)
1176 def record_event(self
, evt
, evt_desc
, evt_details
):
1177 """ Record an event """
1178 self
._log
.debug("Recording event - evt %s, evt_descr %s len = %s",
1179 evt
, evt_desc
, len(self
._events
))
1180 if len(self
._events
) >= NetworkServiceStatus
.MAX_EVENTS_RECORDED
:
1181 self
._events
.popleft()
1182 self
._events
.append((int(time
.time()), evt
, evt_desc
,
1183 evt_details
if evt_details
is not None else None))
1185 self
._loop
.create_task(self
.create_notification(evt
,evt_desc
,evt_details
))
1187 def set_state(self
, state
):
1188 """ set the state of this status object """
1192 """ Return the state as a yang enum string """
1193 state_to_str_map
= {"INIT": "init",
1194 "VL_INIT_PHASE": "vl_init_phase",
1195 "VNF_INIT_PHASE": "vnf_init_phase",
1196 "VNFFG_INIT_PHASE": "vnffg_init_phase",
1197 "SCALING_GROUP_INIT_PHASE": "scaling_group_init_phase",
1198 "RUNNING": "running",
1199 "SCALING_OUT": "scaling_out",
1200 "SCALING_IN": "scaling_in",
1201 "TERMINATE_RCVD": "terminate_rcvd",
1202 "TERMINATE": "terminate",
1203 "VL_TERMINATE_PHASE": "vl_terminate_phase",
1204 "VNF_TERMINATE_PHASE": "vnf_terminate_phase",
1205 "VNFFG_TERMINATE_PHASE": "vnffg_terminate_phase",
1206 "TERMINATED": "terminated",
1208 "VL_INSTANTIATE": "vl_instantiate",
1209 "VL_TERMINATE": "vl_terminate",
1211 return state_to_str_map
[self
._state
.name
]
1215 """ State of this status object """
1220 """ Network Service Record as a message"""
1223 for entry
in self
._events
:
1224 event
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_OperationalEvents()
1227 event
.timestamp
, event
.event
, event
.description
, event
.details
= entry
1228 event_list
.append(event
)
1232 class NetworkServiceRecord(object):
1233 """ Network service record """
1234 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
1236 def __init__(self
, dts
, log
, loop
, nsm
, nsm_plugin
, nsr_cfg_msg
, sdn_account_name
, key_pairs
, restart_mode
=False,
1242 self
._nsr
_cfg
_msg
= nsr_cfg_msg
1243 self
._nsm
_plugin
= nsm_plugin
1244 self
._sdn
_account
_name
= sdn_account_name
1245 self
._vlr
_handler
= vlr_handler
1248 self
._nsr
_msg
= None
1249 self
._nsr
_regh
= None
1250 self
._key
_pairs
= key_pairs
1255 self
._param
_pools
= {}
1256 self
._scaling
_groups
= {}
1257 self
._create
_time
= int(time
.time())
1258 self
._op
_status
= NetworkServiceStatus(dts
, log
, loop
)
1259 self
._config
_status
= NsrYang
.ConfigStates
.CONFIGURING
1260 self
._config
_status
_details
= None
1262 self
.restart_mode
= restart_mode
1263 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
1264 self
._debug
_running
= False
1265 self
._is
_active
= False
1266 self
._vl
_phase
_completed
= False
1267 self
._vnf
_phase
_completed
= False
1270 # Initalise the state to init
1271 # The NSR moves through the following transitions
1272 # 1. INIT -> VLS_READY once all the VLs in the NSD are created
1273 # 2. VLS_READY - VNFS_READY when all the VNFs in the NSD are created
1274 # 3. VNFS_READY - READY when the NSR is published
1276 self
.set_state(NetworkServiceRecordState
.INIT
)
1278 self
.substitute_input_parameters
= InputParameterSubstitution(self
._log
)
1281 def nsm_plugin(self
):
1283 return self
._nsm
_plugin
1285 def set_state(self
, state
):
1286 """ Set state for this NSR"""
1287 self
._log
.debug("Setting state to %s", state
)
1288 # We are in init phase and is moving to the next state
1289 # The new state could be a FAILED state or VNF_INIIT_PHASE
1290 if self
.state
== NetworkServiceRecordState
.VL_INIT_PHASE
:
1291 self
._vl
_phase
_completed
= True
1293 if self
.state
== NetworkServiceRecordState
.VNF_INIT_PHASE
:
1294 self
._vnf
_phase
_completed
= True
1296 self
._op
_status
.set_state(state
)
1300 """ Get id for this NSR"""
1301 return self
._nsr
_cfg
_msg
.id
1305 """ Name of this network service record """
1306 return self
._nsr
_cfg
_msg
.name
1309 def cloud_account_name(self
):
1310 return self
._nsr
_cfg
_msg
.cloud_account
1313 def om_datacenter_name(self
):
1314 if self
._nsr
_cfg
_msg
.has_field('om_datacenter'):
1315 return self
._nsr
_cfg
_msg
.om_datacenter
1320 """State of this NetworkServiceRecord"""
1321 return self
._op
_status
.state
1325 """ Is this NSR active ?"""
1326 return True if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
else False
1330 """ VLRs associated with this NSR"""
1335 """ VNFRs associated with this NSR"""
1340 """ VNFFGRs associated with this NSR"""
1341 return self
._vnffgrs
1344 def scaling_groups(self
):
1345 """ Scaling groups associated with this NSR """
1346 return self
._scaling
_groups
1349 def param_pools(self
):
1350 """ Parameter value pools associated with this NSR"""
1351 return self
._param
_pools
1354 def nsr_cfg_msg(self
):
1355 return self
._nsr
_cfg
_msg
1358 def nsr_cfg_msg(self
, msg
):
1359 self
._nsr
_cfg
_msg
= msg
1363 """ NSD Protobuf for this NSR """
1364 if self
._nsd
is not None:
1366 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
1371 """ NSD ID for this NSR """
1372 return self
.nsd_msg
.id
1376 ''' Get a new job id for config primitive'''
1381 def config_status(self
):
1382 """ Config status for NSR """
1383 return self
._config
_status
1385 def resolve_placement_group_cloud_construct(self
, input_group
):
1387 Returns the cloud specific construct for placement group
1389 copy_dict
= ['name', 'requirement', 'strategy']
1391 for group_info
in self
._nsr
_cfg
_msg
.nsd_placement_group_maps
:
1392 if group_info
.placement_group_ref
== input_group
.name
:
1393 group
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_PlacementGroupsInfo()
1394 group_dict
= {k
:v
for k
,v
in
1395 group_info
.as_dict().items() if k
!= 'placement_group_ref'}
1396 for param
in copy_dict
:
1397 group_dict
.update({param
: getattr(input_group
, param
)})
1398 group
.from_dict(group_dict
)
1404 return "NSR(name={}, nsd_id={}, cloud_account={})".format(
1405 self
.name
, self
.nsd_id
, self
.cloud_account_name
1408 def _get_vnfd(self
, vnfd_id
, config_xact
):
1409 """ Fetch vnfd msg for the passed vnfd id """
1410 return self
._nsm
.get_vnfd(vnfd_id
, config_xact
)
1412 def _get_vnfd_cloud_account(self
, vnfd_member_index
):
1413 """ Fetch Cloud Account for the passed vnfd id """
1414 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1415 vim_accounts
= [(vnf
.cloud_account
,vnf
.om_datacenter
) for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map \
1416 if vnfd_member_index
== vnf
.member_vnf_index_ref
]
1417 if vim_accounts
and vim_accounts
[0]:
1418 return vim_accounts
[0]
1419 return (self
.cloud_account_name
,self
.om_datacenter_name
)
1421 def _get_constituent_vnfd_msg(self
, vnf_index
):
1422 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1423 if const_vnfd
.member_vnf_index
== vnf_index
:
1426 raise ValueError("Constituent VNF index %s not found" % vnf_index
)
1428 def record_event(self
, evt
, evt_desc
, evt_details
=None, state
=None):
1429 """ Record an event """
1430 self
._op
_status
.record_event(evt
, evt_desc
, evt_details
)
1431 if state
is not None:
1432 self
.set_state(state
)
1434 def scaling_trigger_str(self
, trigger
):
1435 SCALING_TRIGGER_STRS
= {
1436 NsdYang
.ScalingTrigger
.PRE_SCALE_IN
: 'pre-scale-in',
1437 NsdYang
.ScalingTrigger
.POST_SCALE_IN
: 'post-scale-in',
1438 NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
: 'pre-scale-out',
1439 NsdYang
.ScalingTrigger
.POST_SCALE_OUT
: 'post-scale-out',
1442 return SCALING_TRIGGER_STRS
[trigger
]
1443 except Exception as e
:
1444 self
._log
.error("Scaling trigger mapping error for {} : {}".
1446 self
._log
.exception(e
)
1447 return "Unknown trigger"
1450 def instantiate_vls(self
):
1452 This function instantiates VLs for every VL in this Network Service
1454 self
._log
.debug("Instantiating %d VLs in NSD id %s", len(self
._vlrs
),
1456 for vlr
in self
._vlrs
:
1457 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1458 vlr
.state
= VlRecordState
.ACTIVE
1459 self
._loop
.create_task(self
.vlr_uptime_update(vlr
))
1462 def vlr_uptime_update(self
, vlr
):
1463 vlr_
= RwVlrYang
.YangData_Vlr_VlrCatalog_Vlr
.from_dict({'id': vlr
.id})
1465 if vlr
.state
not in [VlRecordState
.INIT
, VlRecordState
.INSTANTIATION_PENDING
, VlRecordState
.ACTIVE
]:
1467 vlr_
.uptime
= int(time
.time()) - vlr
._create
_time
1468 yield from self
._vlr
_handler
.update(None, VirtualLinkRecord
.vlr_xpath(vlr
), vlr_
)
1469 yield from asyncio
.sleep(2, loop
=self
._loop
)
1473 def create(self
, config_xact
):
1474 """ Create this network service"""
1475 # Create virtual links for all the external vnf
1476 # connection points in this NS
1477 yield from self
.create_vls()
1479 # Create VNFs in this network service
1480 yield from self
.create_vnfs(config_xact
)
1482 # Create VNFFG for network service
1483 self
.create_vnffgs()
1485 # Create Scaling Groups for each scaling group in NSD
1486 self
.create_scaling_groups()
1488 # Create Parameter Pools
1489 self
.create_param_pools()
1492 def apply_scale_group_config_script(self
, script
, group
, scale_instance
, trigger
, vnfrs
=None):
1493 """ Apply config based on script for scale group """
1496 def add_vnfrs_data(vnfrs_list
):
1497 """ Add as a dict each of the VNFRs data """
1499 for vnfr
in vnfrs_list
:
1500 self
._log
.debug("Add VNFR {} data".format(vnfr
))
1502 vnfr_data
['name'] = vnfr
.name
1503 if trigger
in [NsdYang
.ScalingTrigger
.PRE_SCALE_IN
, NsdYang
.ScalingTrigger
.POST_SCALE_OUT
]:
1504 # Get VNF management and other IPs, etc
1505 opdata
= yield from self
.fetch_vnfr(vnfr
.xpath
)
1506 self
._log
.debug("VNFR {} op data: {}".format(vnfr
.name
, opdata
))
1508 vnfr_data
['rw_mgmt_ip'] = opdata
.mgmt_interface
.ip_address
1509 vnfr_data
['rw_mgmt_port'] = opdata
.mgmt_interface
.port
1510 except Exception as e
:
1511 self
._log
.error("Unable to get management IP for vnfr {}:{}".
1512 format(vnfr
.name
, e
))
1515 vnfr_data
['connection_points'] = []
1516 for cp
in opdata
.connection_point
:
1518 con_pt
['name'] = cp
.name
1519 con_pt
['ip_address'] = cp
.ip_address
1520 vnfr_data
['connection_points'].append(con_pt
)
1521 except Exception as e
:
1522 self
._log
.error("Exception getting connections points for VNFR {}: {}".
1523 format(vnfr
.name
, e
))
1525 vnfrs_data
.append(vnfr_data
)
1526 self
._log
.debug("VNFRs data: {}".format(vnfrs_data
))
1530 def add_nsr_data(nsr
):
1532 nsr_data
['name'] = nsr
.name
1535 if script
is None or len(script
) == 0:
1536 self
._log
.error("Script not provided for scale group config: {}".format(group
.name
))
1539 if script
[0] == '/':
1542 path
= os
.path
.join(os
.environ
['RIFT_INSTALL'], "usr/bin", script
)
1543 if not os
.path
.exists(path
):
1544 self
._log
.error("Config faled for scale group {}: Script does not exist at {}".
1545 format(group
.name
, path
))
1548 # Build a YAML file with all parameters for the script to execute
1549 # The data consists of 5 sections
1551 # 2. Scale group config
1552 # 3. VNFRs in the scale group
1553 # 4. VNFRs outside scale group
1556 data
['trigger'] = group
.trigger_map(trigger
)
1557 data
['config'] = group
.group_msg
.as_dict()
1560 data
["vnfrs_in_group"] = yield from add_vnfrs_data(vnfrs
)
1562 data
["vnfrs_in_group"] = yield from add_vnfrs_data(scale_instance
.vnfrs
)
1564 data
["vnfrs_others"] = yield from add_vnfrs_data(self
.vnfrs
.values())
1565 data
["nsr"] = add_nsr_data(self
)
1568 with tempfile
.NamedTemporaryFile(delete
=False) as tmp_file
:
1569 tmp_file
.write(yaml
.dump(data
, default_flow_style
=True)
1572 self
._log
.debug("Creating a temp file: {} with input data: {}".
1573 format(tmp_file
.name
, data
))
1575 cmd
= "{} {}".format(path
, tmp_file
.name
)
1576 self
._log
.debug("Running the CMD: {}".format(cmd
))
1577 proc
= yield from asyncio
.create_subprocess_shell(cmd
, loop
=self
._loop
)
1578 rc
= yield from proc
.wait()
1580 self
._log
.error("The script {} for scale group {} config returned: {}".
1581 format(script
, group
.name
, rc
))
1589 def apply_scaling_group_config(self
, trigger
, group
, scale_instance
, vnfrs
=None):
1590 """ Apply the config for the scaling group based on trigger """
1591 if group
is None or scale_instance
is None:
1595 def update_config_status(success
=True, err_msg
=None):
1596 self
._log
.debug("Update %s config status to %r : %s",
1597 scale_instance
, success
, err_msg
)
1598 if (scale_instance
.config_status
== "failed"):
1599 # Do not update the config status if it is already in failed state
1602 if scale_instance
.config_status
== "configured":
1603 # Update only to failed state an already configured scale instance
1605 scale_instance
.config_status
= "failed"
1606 scale_instance
.config_err_msg
= err_msg
1607 yield from self
.update_state()
1609 # We are in configuring state
1610 # Only after post scale out mark instance as configured
1611 if trigger
== NsdYang
.ScalingTrigger
.POST_SCALE_OUT
:
1613 scale_instance
.config_status
= "configured"
1615 scale_instance
.config_status
= "failed"
1616 scale_instance
.config_err_msg
= err_msg
1617 yield from self
.update_state()
1619 config
= group
.trigger_config(trigger
)
1623 self
._log
.debug("Scaling group {} config: {}".format(group
.name
, config
))
1624 if config
.has_field("ns_config_primitive_name_ref"):
1625 config_name
= config
.ns_config_primitive_name_ref
1626 nsd_msg
= self
.nsd_msg
1627 config_primitive
= None
1628 for ns_cfg_prim
in nsd_msg
.service_primitive
:
1629 if ns_cfg_prim
.name
== config_name
:
1630 config_primitive
= ns_cfg_prim
1633 if config_primitive
is None:
1634 raise ValueError("Could not find ns_cfg_prim %s in nsr %s" % (config_name
, self
.name
))
1636 self
._log
.debug("Scaling group {} config primitive: {}".format(group
.name
, config_primitive
))
1637 if config_primitive
.has_field("user_defined_script"):
1638 rc
= yield from self
.apply_scale_group_config_script(config_primitive
.user_defined_script
,
1639 group
, scale_instance
, trigger
, vnfrs
)
1642 err_msg
= "Failed config for trigger {} using config script '{}'". \
1643 format(self
.scaling_trigger_str(trigger
),
1644 config_primitive
.user_defined_script
)
1645 yield from update_config_status(success
=rc
, err_msg
=err_msg
)
1648 err_msg
= "Failed config for trigger {} as config script is not specified". \
1649 format(self
.scaling_trigger_str(trigger
))
1650 yield from update_config_status(success
=False, err_msg
=err_msg
)
1651 raise NotImplementedError("Only script based config support for scale group for now: {}".
1654 err_msg
= "Failed config for trigger {} as config primitive is not specified".\
1655 format(self
.scaling_trigger_str(trigger
))
1656 yield from update_config_status(success
=False, err_msg
=err_msg
)
1657 self
._log
.error("Config primitive not specified for config action in scale group %s" %
1661 def create_scaling_groups(self
):
1662 """ This function creates a NSScalingGroup for every scaling
1663 group defined in he NSD"""
1665 for scaling_group_msg
in self
.nsd_msg
.scaling_group_descriptor
:
1666 self
._log
.debug("Found scaling_group %s in nsr id %s",
1667 scaling_group_msg
.name
, self
.id)
1669 group_record
= scale_group
.ScalingGroup(
1674 self
._scaling
_groups
[group_record
.name
] = group_record
1677 def create_scale_group_instance(self
, group_name
, index
, config_xact
, is_default
=False):
1678 group
= self
._scaling
_groups
[group_name
]
1679 scale_instance
= group
.create_instance(index
, is_default
)
1683 self
._log
.debug("Creating %u VNFs associated with NS id %s scaling group %s",
1684 len(self
.nsd_msg
.constituent_vnfd
), self
.id, self
)
1687 for vnf_index
, count
in group
.vnf_index_count_map
.items():
1688 const_vnfd_msg
= self
._get
_constituent
_vnfd
_msg
(vnf_index
)
1689 vnfd_msg
= self
._get
_vnfd
(const_vnfd_msg
.vnfd_id_ref
, config_xact
)
1691 cloud_account_name
, om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd_msg
.member_vnf_index
)
1692 if cloud_account_name
is None:
1693 cloud_account_name
= self
.cloud_account_name
1694 for _
in range(count
):
1695 vnfr
= yield from self
.create_vnf_record(vnfd_msg
, const_vnfd_msg
, cloud_account_name
, om_datacenter_name
, group_name
, index
)
1696 scale_instance
.add_vnfr(vnfr
)
1702 def instantiate_instance():
1703 self
._log
.debug("Creating %s VNFRS", scale_instance
)
1704 vnfrs
= yield from create_vnfs()
1705 yield from self
.publish()
1707 self
._log
.debug("Instantiating %s VNFRS for %s", len(vnfrs
), scale_instance
)
1708 scale_instance
.operational_status
= "vnf_init_phase"
1709 yield from self
.update_state()
1712 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
,
1713 group
, scale_instance
, vnfrs
)
1715 self
._log
.error("Pre scale out config for scale group {} ({}) failed".
1716 format(group
.name
, index
))
1717 scale_instance
.operational_status
= "failed"
1719 yield from self
.instantiate_vnfs(vnfrs
)
1721 except Exception as e
:
1722 self
._log
.exception("Failed to begin instantiatiation of vnfs for scale group {}: {}".
1723 format(group
.name
, e
))
1724 self
._log
.exception(e
)
1725 scale_instance
.operational_status
= "failed"
1727 yield from self
.update_state()
1729 yield from instantiate_instance()
1732 def delete_scale_group_instance(self
, group_name
, index
):
1733 group
= self
._scaling
_groups
[group_name
]
1734 scale_instance
= group
.get_instance(index
)
1735 if scale_instance
.is_default
:
1736 raise ScalingOperationError("Cannot terminate a default scaling group instance")
1738 scale_instance
.operational_status
= "terminate"
1739 yield from self
.update_state()
1742 def terminate_instance():
1743 self
._log
.debug("Terminating %s VNFRS" % scale_instance
)
1744 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_IN
,
1745 group
, scale_instance
)
1747 self
._log
.error("Pre scale in config for scale group {} ({}) failed".
1748 format(group
.name
, index
))
1750 # Going ahead with terminate, even if there is an error in pre-scale-in config
1751 # as this could be result of scale out failure and we need to cleanup this group
1752 yield from self
.terminate_vnfrs(scale_instance
.vnfrs
)
1753 group
.delete_instance(index
)
1755 scale_instance
.operational_status
= "vnf_terminate_phase"
1756 yield from self
.update_state()
1758 yield from terminate_instance()
1761 def _update_scale_group_instances_status(self
):
1763 def post_scale_out_task(group
, instance
):
1764 # Apply post scale out config once all VNFRs are active
1765 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_OUT
,
1767 instance
.operational_status
= "running"
1769 self
._log
.debug("Scale out for group {} and instance {} succeeded".
1770 format(group
.name
, instance
.instance_id
))
1772 self
._log
.error("Post scale out config for scale group {} ({}) failed".
1773 format(group
.name
, instance
.instance_id
))
1775 yield from self
.update_state()
1777 group_instances
= {group
: group
.instances
for group
in self
._scaling
_groups
.values()}
1778 for group
, instances
in group_instances
.items():
1779 self
._log
.debug("Updating %s instance status", group
)
1780 for instance
in instances
:
1781 instance_vnf_state_list
= [vnfr
.state
for vnfr
in instance
.vnfrs
]
1782 self
._log
.debug("Got vnfr instance states: %s", instance_vnf_state_list
)
1783 if instance
.operational_status
== "vnf_init_phase":
1784 if all([state
== VnfRecordState
.ACTIVE
for state
in instance_vnf_state_list
]):
1785 instance
.operational_status
= "running"
1787 # Create a task for post scale out to allow us to sleep before attempting
1788 # to configure newly created VM's
1789 self
._loop
.create_task(post_scale_out_task(group
, instance
))
1791 elif any([state
== VnfRecordState
.FAILED
for state
in instance_vnf_state_list
]):
1792 self
._log
.debug("Scale out for group {} and instance {} failed".
1793 format(group
.name
, instance
.instance_id
))
1794 instance
.operational_status
= "failed"
1796 elif instance
.operational_status
== "vnf_terminate_phase":
1797 if all([state
== VnfRecordState
.TERMINATED
for state
in instance_vnf_state_list
]):
1798 instance
.operational_status
= "terminated"
1799 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_IN
,
1802 self
._log
.debug("Scale in for group {} and instance {} succeeded".
1803 format(group
.name
, instance
.instance_id
))
1805 self
._log
.error("Post scale in config for scale group {} ({}) failed".
1806 format(group
.name
, instance
.instance_id
))
1808 def create_vnffgs(self
):
1809 """ This function creates VNFFGs for every VNFFG in the NSD
1810 associated with this NSR"""
1812 for vnffgd
in self
.nsd_msg
.vnffgd
:
1813 self
._log
.debug("Found vnffgd %s in nsr id %s", vnffgd
, self
.id)
1814 vnffgr
= VnffgRecord(self
._dts
,
1817 self
._nsm
._vnffgmgr
,
1821 self
._sdn
_account
_name
1823 self
._vnffgrs
[vnffgr
.id] = vnffgr
1825 def resolve_vld_ip_profile(self
, nsd_msg
, vld
):
1826 self
._log
.debug("Receieved ip profile ref is %s",vld
.ip_profile_ref
)
1827 if not vld
.has_field('ip_profile_ref'):
1829 profile
= [profile
for profile
in nsd_msg
.ip_profiles
if profile
.name
== vld
.ip_profile_ref
]
1830 return profile
[0] if profile
else None
1833 def _create_vls(self
, vld
, cloud_account
,om_datacenter
):
1834 """Create a VLR in the cloud account specified using the given VLD
1838 cloud_account : Cloud account name
1843 vlr
= yield from VirtualLinkRecord
.create_record(
1851 self
.resolve_vld_ip_profile(self
.nsd_msg
, vld
),
1853 restart_mode
=self
.restart_mode
)
1857 def _extract_cloud_accounts_for_vl(self
, vld
):
1859 Extracts the list of cloud accounts from the NS Config obj
1862 1. Cloud accounts based connection point (vnf_cloud_account_map)
1864 vld : VLD yang object
1869 cloud_account_list
= []
1871 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1872 # Handle case where cloud_account is None
1874 for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1875 if vnf
.cloud_account
is not None or vnf
.om_datacenter
is not None:
1876 vnf_cloud_map
[vnf
.member_vnf_index_ref
] = (vnf
.cloud_account
,vnf
.om_datacenter
)
1878 for vnfc
in vld
.vnfd_connection_point_ref
:
1879 cloud_account
= vnf_cloud_map
.get(
1880 vnfc
.member_vnf_index_ref
,
1881 (self
.cloud_account_name
,self
.om_datacenter_name
))
1883 cloud_account_list
.append(cloud_account
)
1885 if self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1886 for vld_map
in self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1887 if vld_map
.vld_id_ref
== vld
.id:
1888 for cloud_account
in vld_map
.cloud_accounts
:
1889 cloud_account_list
.extend((cloud_account
,None))
1890 for om_datacenter
in vld_map
.om_datacenters
:
1891 cloud_account_list
.extend((None,om_datacenter
))
1893 # If no config has been provided then fall-back to the default
1895 if not cloud_account_list
:
1896 cloud_account_list
= [(self
.cloud_account_name
,self
.om_datacenter_name
)]
1898 self
._log
.debug("VL {} cloud accounts: {}".
1899 format(vld
.name
, cloud_account_list
))
1900 return set(cloud_account_list
)
1903 def create_vls(self
):
1904 """ This function creates VLs for every VLD in the NSD
1905 associated with this NSR"""
1906 for vld
in self
.nsd_msg
.vld
:
1907 self
._log
.debug("Found vld %s in nsr id %s", vld
, self
.id)
1908 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1909 for cloud_account
,om_datacenter
in cloud_account_list
:
1910 vlr
= yield from self
._create
_vls
(vld
, cloud_account
,om_datacenter
)
1911 self
._vlrs
.append(vlr
)
1915 def create_vl_instance(self
, vld
):
1916 self
._log
.debug("Create VL for {}: {}".format(self
.id, vld
.as_dict()))
1917 # Check if the VL is already present
1919 for vl
in self
._vlrs
:
1920 if vl
.vld_msg
.id == vld
.id:
1921 self
._log
.debug("The VLD %s already in NSR %s as VLR %s with status %s",
1922 vld
.id, self
.id, vl
.id, vl
.state
)
1924 if vlr
.state
!= VlRecordState
.TERMINATED
:
1925 err_msg
= "VLR for VL %s in NSR %s already instantiated", \
1927 self
._log
.error(err_msg
)
1928 raise NsrVlUpdateError(err_msg
)
1932 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1933 for account
,om_datacenter
in cloud_account_list
:
1934 vlr
= yield from self
._create
_vls
(vld
, account
,om_datacenter
)
1935 self
._vlrs
.append(vlr
)
1937 vlr
.state
= VlRecordState
.INSTANTIATION_PENDING
1938 yield from self
.update_state()
1941 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1942 vlr
.state
= VlRecordState
.ACTIVE
1944 except Exception as e
:
1945 err_msg
= "Error instantiating VL for NSR {} and VLD {}: {}". \
1946 format(self
.id, vld
.id, e
)
1947 self
._log
.error(err_msg
)
1948 self
._log
.exception(e
)
1949 vlr
.state
= VlRecordState
.FAILED
1951 yield from self
.update_state()
1954 def delete_vl_instance(self
, vld
):
1955 for vlr
in self
._vlrs
:
1956 if vlr
.vld_msg
.id == vld
.id:
1957 self
._log
.debug("Found VLR %s for VLD %s in NSR %s",
1958 vlr
.id, vld
.id, self
.id)
1959 vlr
.state
= VlRecordState
.TERMINATE_PENDING
1960 yield from self
.update_state()
1963 yield from self
.nsm_plugin
.terminate_vl(vlr
)
1964 vlr
.state
= VlRecordState
.TERMINATED
1965 self
._vlrs
.remove(vlr
)
1967 except Exception as e
:
1968 err_msg
= "Error terminating VL for NSR {} and VLD {}: {}". \
1969 format(self
.id, vld
.id, e
)
1970 self
._log
.error(err_msg
)
1971 self
._log
.exception(e
)
1972 vlr
.state
= VlRecordState
.FAILED
1974 yield from self
.update_state()
1978 def create_vnfs(self
, config_xact
):
1980 This function creates VNFs for every VNF in the NSD
1981 associated with this NSR
1983 self
._log
.debug("Creating %u VNFs associated with this NS id %s",
1984 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
1986 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1987 if not const_vnfd
.start_by_default
:
1988 self
._log
.debug("start_by_default set to False in constituent VNF (%s). Skipping start.",
1989 const_vnfd
.member_vnf_index
)
1992 vnfd_msg
= self
._get
_vnfd
(const_vnfd
.vnfd_id_ref
, config_xact
)
1993 cloud_account_name
,om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd
.member_vnf_index
)
1994 if cloud_account_name
is None:
1995 cloud_account_name
= self
.cloud_account_name
1996 yield from self
.create_vnf_record(vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
)
1999 def get_placement_groups(self
, vnfd_msg
, const_vnfd
):
2000 placement_groups
= []
2001 for group
in self
.nsd_msg
.placement_groups
:
2002 for member_vnfd
in group
.member_vnfd
:
2003 if (member_vnfd
.vnfd_id_ref
== vnfd_msg
.id) and \
2004 (member_vnfd
.member_vnf_index_ref
== const_vnfd
.member_vnf_index
):
2005 group_info
= self
.resolve_placement_group_cloud_construct(group
)
2006 if group_info
is None:
2007 self
._log
.error("Could not resolve cloud-construct for placement group: %s", group
.name
)
2008 ### raise PlacementGroupError("Could not resolve cloud-construct for placement group: {}".format(group.name))
2010 self
._log
.info("Successfully resolved cloud construct for placement group: %s for VNF: %s (Member Index: %s)",
2013 const_vnfd
.member_vnf_index
)
2014 placement_groups
.append(group_info
)
2015 return placement_groups
2018 def create_vnf_record(self
, vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
, group_name
=None, group_instance_id
=None):
2019 # Fetch the VNFD associated with this VNF
2020 placement_groups
= self
.get_placement_groups(vnfd_msg
, const_vnfd
)
2021 self
._log
.info("Cloud Account for VNF %d is %s",const_vnfd
.member_vnf_index
,cloud_account_name
)
2022 self
._log
.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s",
2024 const_vnfd
.member_vnf_index
,
2025 [ group
.name
for group
in placement_groups
])
2026 vnfr
= yield from VirtualNetworkFunctionRecord
.create_record(self
._dts
,
2039 restart_mode
=self
.restart_mode
,
2041 if vnfr
.id in self
._vnfrs
:
2042 err
= "VNF with VNFR id %s already in vnf list" % (vnfr
.id,)
2043 raise NetworkServiceRecordError(err
)
2045 self
._vnfrs
[vnfr
.id] = vnfr
2046 self
._nsm
.vnfrs
[vnfr
.id] = vnfr
2048 yield from vnfr
.set_config_status(NsrYang
.ConfigStates
.INIT
)
2050 self
._log
.debug("Added VNFR %s to NSM VNFR list with id %s",
2056 def create_param_pools(self
):
2057 for param_pool
in self
.nsd_msg
.parameter_pool
:
2058 self
._log
.debug("Found parameter pool %s in nsr id %s", param_pool
, self
.id)
2060 start_value
= param_pool
.range.start_value
2061 end_value
= param_pool
.range.end_value
2062 if end_value
< start_value
:
2063 raise NetworkServiceRecordError(
2064 "Parameter pool %s has invalid range (start: {}, end: {})".format(
2065 start_value
, end_value
2069 self
._param
_pools
[param_pool
.name
] = config_value_pool
.ParameterValuePool(
2072 range(start_value
, end_value
)
2076 def fetch_vnfr(self
, vnfr_path
):
2077 """ Fetch VNFR record """
2079 self
._log
.debug("Fetching VNFR with key %s while instantiating %s",
2081 res_iter
= yield from self
._dts
.query_read(vnfr_path
, rwdts
.XactFlag
.MERGE
)
2083 for ent
in res_iter
:
2084 res
= yield from ent
2090 def instantiate_vnfs(self
, vnfrs
):
2092 This function instantiates VNFs for every VNF in this Network Service
2094 self
._log
.debug("Instantiating %u VNFs in NS %s", len(vnfrs
), self
.id)
2096 self
._log
.debug("Instantiating VNF: %s in NS %s", vnf
, self
.id)
2097 yield from self
.nsm_plugin
.instantiate_vnf(self
, vnf
)
2100 def instantiate_vnffgs(self
):
2102 This function instantiates VNFFGs for every VNFFG in this Network Service
2104 self
._log
.debug("Instantiating %u VNFFGs in NS %s",
2105 len(self
.nsd_msg
.vnffgd
), self
.id)
2106 for _
, vnfr
in self
.vnfrs
.items():
2107 while vnfr
.state
in [VnfRecordState
.INSTANTIATION_PENDING
, VnfRecordState
.INIT
]:
2108 self
._log
.debug("Received vnfr state for vnfr %s is %s; retrying",vnfr
.name
,vnfr
.state
)
2109 yield from asyncio
.sleep(2, loop
=self
._loop
)
2110 if vnfr
.state
== VnfRecordState
.ACTIVE
:
2111 self
._log
.debug("Received vnfr state for vnfr %s is %s ",vnfr
.name
,vnfr
.state
)
2114 self
._log
.debug("Received vnfr state for vnfr %s is %s; failing vnffg creation",vnfr
.name
,vnfr
.state
)
2115 self
._vnffgr
_state
= VnffgRecordState
.FAILED
2118 self
._log
.info("Waiting for 90 seconds for VMs to come up")
2119 yield from asyncio
.sleep(90, loop
=self
._loop
)
2120 self
._log
.info("Starting VNFFG orchestration")
2121 for vnffg
in self
._vnffgrs
.values():
2122 self
._log
.debug("Instantiating VNFFG: %s in NS %s", vnffg
, self
.id)
2123 yield from vnffg
.instantiate()
2126 def instantiate_scaling_instances(self
, config_xact
):
2127 """ Instantiate any default scaling instances in this Network Service """
2128 for group
in self
._scaling
_groups
.values():
2129 for i
in range(group
.min_instance_count
):
2130 self
._log
.debug("Instantiating %s default scaling instance %s", group
, i
)
2131 yield from self
.create_scale_group_instance(
2132 group
.name
, i
, config_xact
, is_default
=True
2135 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2136 if group_msg
.scaling_group_name_ref
!= group
.name
:
2139 for instance
in group_msg
.instance
:
2140 self
._log
.debug("Reloading %s scaling instance %s", group_msg
, instance
.id)
2141 yield from self
.create_scale_group_instance(
2142 group
.name
, instance
.id, config_xact
, is_default
=False
2145 def has_scaling_instances(self
):
2146 """ Return boolean indicating if the network service has default scaling groups """
2147 for group
in self
._scaling
_groups
.values():
2148 if group
.min_instance_count
> 0:
2151 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2152 if len(group_msg
.instance
) > 0:
2159 """ This function publishes this NSR """
2160 self
._nsr
_msg
= self
.create_msg()
2162 self
._log
.debug("Publishing the NSR with xpath %s and nsr %s",
2166 if self
._debug
_running
:
2167 self
._log
.debug("Publishing NSR in RUNNING state!")
2170 with self
._dts
.transaction() as xact
:
2171 yield from self
._nsm
.nsr_handler
.update(xact
, self
.nsr_xpath
, self
._nsr
_msg
)
2172 if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
:
2173 self
._debug
_running
= True
2176 def unpublish(self
, xact
):
2177 """ Unpublish this NSR object """
2178 self
._log
.debug("Unpublishing Network service id %s", self
.id)
2179 yield from self
._nsm
.nsr_handler
.delete(xact
, self
.nsr_xpath
)
2182 def nsr_xpath(self
):
2183 """ Returns the xpath associated with this NSR """
2185 "D,/nsr:ns-instance-opdata" +
2186 "/nsr:nsr[nsr:ns-instance-config-ref = '{}']"
2190 def xpath_from_nsr(nsr
):
2191 """ Returns the xpath associated with this NSR op data"""
2192 return (NetworkServiceRecord
.XPATH
+
2193 "[nsr:ns-instance-config-ref = '{}']").format(nsr
.id)
2196 def nsd_xpath(self
):
2197 """ Return NSD config xpath."""
2199 "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}']"
2200 ).format(self
.nsd_id
)
2203 def instantiate(self
, config_xact
):
2204 """"Instantiates a NetworkServiceRecord.
2206 This function instantiates a Network service
2207 which involves the following steps,
2209 * Instantiate every VL in NSD by sending create VLR request to DTS.
2210 * Instantiate every VNF in NSD by sending create VNF reuqest to DTS.
2211 * Publish the NSR details to DTS
2214 nsr: The NSR configuration request containing nsr-id and nsd
2215 config_xact: The configuration transaction which initiated the instatiation
2218 NetworkServiceRecordError if the NSR creation fails
2224 self
._log
.debug("Instantiating NS - %s xact - %s", self
, config_xact
)
2226 # Move the state to INIITALIZING
2227 self
.set_state(NetworkServiceRecordState
.INIT
)
2229 event_descr
= "Instantiation Request Received NSR Id:%s" % self
.id
2230 self
.record_event("instantiating", event_descr
)
2233 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
2236 # Update ref count if nsd present in catalog
2237 self
._nsm
.get_nsd_ref(self
.nsd_id
)
2239 except NetworkServiceDescriptorError
:
2240 # This could be an NSD not in the nsd-catalog
2243 # Merge any config and initial config primitive values
2244 self
.config_store
.merge_nsd_config(self
.nsd_msg
)
2245 self
._log
.debug("Merged NSD: {}".format(self
.nsd_msg
.as_dict()))
2247 event_descr
= "Fetched NSD with descriptor id %s" % self
.nsd_id
2248 self
.record_event("nsd-fetched", event_descr
)
2250 if self
._nsd
is None:
2251 msg
= "Failed to fetch NSD with nsd-id [%s] for nsr-id %s"
2252 self
._log
.debug(msg
, self
.nsd_id
, self
.id)
2253 raise NetworkServiceRecordError(self
)
2255 self
._log
.debug("Got nsd result %s", self
._nsd
)
2257 # Substitute any input parameters
2258 self
.substitute_input_parameters(self
._nsd
, self
._nsr
_cfg
_msg
)
2261 yield from self
.create(config_xact
)
2263 # Publish the NSR to DTS
2264 yield from self
.publish()
2267 def do_instantiate():
2269 Instantiate network service
2271 self
._log
.debug("Instantiating VLs nsr id [%s] nsd id [%s]",
2272 self
.id, self
.nsd_id
)
2274 # instantiate the VLs
2275 event_descr
= ("Instantiating %s external VLs for NSR id %s" %
2276 (len(self
.nsd_msg
.vld
), self
.id))
2277 self
.record_event("begin-external-vls-instantiation", event_descr
)
2279 self
.set_state(NetworkServiceRecordState
.VL_INIT_PHASE
)
2281 yield from self
.instantiate_vls()
2283 # Publish the NSR to DTS
2284 yield from self
.publish()
2286 event_descr
= ("Finished instantiating %s external VLs for NSR id %s" %
2287 (len(self
.nsd_msg
.vld
), self
.id))
2288 self
.record_event("end-external-vls-instantiation", event_descr
)
2290 self
.set_state(NetworkServiceRecordState
.VNF_INIT_PHASE
)
2292 self
._log
.debug("Instantiating VNFs ...... nsr[%s], nsd[%s]",
2293 self
.id, self
.nsd_id
)
2295 # instantiate the VNFs
2296 event_descr
= ("Instantiating %s VNFS for NSR id %s" %
2297 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2299 self
.record_event("begin-vnf-instantiation", event_descr
)
2301 yield from self
.instantiate_vnfs(self
._vnfrs
.values())
2303 self
._log
.debug(" Finished instantiating %d VNFs for NSR id %s",
2304 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
2306 event_descr
= ("Finished instantiating %s VNFs for NSR id %s" %
2307 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2308 self
.record_event("end-vnf-instantiation", event_descr
)
2310 if len(self
.vnffgrs
) > 0:
2311 #self.set_state(NetworkServiceRecordState.VNFFG_INIT_PHASE)
2312 event_descr
= ("Instantiating %s VNFFGS for NSR id %s" %
2313 (len(self
.nsd_msg
.vnffgd
), self
.id))
2315 self
.record_event("begin-vnffg-instantiation", event_descr
)
2317 yield from self
.instantiate_vnffgs()
2319 event_descr
= ("Finished instantiating %s VNFFGDs for NSR id %s" %
2320 (len(self
.nsd_msg
.vnffgd
), self
.id))
2321 self
.record_event("end-vnffg-instantiation", event_descr
)
2323 if self
.has_scaling_instances():
2324 event_descr
= ("Instantiating %s Scaling Groups for NSR id %s" %
2325 (len(self
._scaling
_groups
), self
.id))
2327 self
.record_event("begin-scaling-group-instantiation", event_descr
)
2328 yield from self
.instantiate_scaling_instances(config_xact
)
2329 self
.record_event("end-scaling-group-instantiation", event_descr
)
2331 # Give the plugin a chance to deploy the network service now that all
2332 # virtual links and vnfs are instantiated
2333 yield from self
.nsm_plugin
.deploy(self
._nsr
_msg
)
2335 self
._log
.debug("Publishing NSR...... nsr[%s], nsd[%s]",
2336 self
.id, self
.nsd_id
)
2338 # Publish the NSR to DTS
2339 yield from self
.publish()
2341 self
._log
.debug("Published NSR...... nsr[%s], nsd[%s]",
2342 self
.id, self
.nsd_id
)
2344 def on_instantiate_done(fut
):
2345 # If the do_instantiate fails, then publish NSR with failed result
2346 if fut
.exception() is not None:
2347 self
._log
.error("NSR instantiation failed for NSR id %s: %s", self
.id, str(fut
.exception()))
2348 self
._loop
.create_task(self
.instantiation_failed(failed_reason
=str(fut
.exception())))
2350 instantiate_task
= self
._loop
.create_task(do_instantiate())
2351 instantiate_task
.add_done_callback(on_instantiate_done
)
2354 def set_config_status(self
, status
, status_details
=None):
2355 if self
.config_status
!= status
:
2356 self
._log
.debug("Updating NSR {} status for {} to {}".
2357 format(self
.name
, self
.config_status
, status
))
2358 self
._config
_status
= status
2359 self
._config
_status
_details
= status_details
2361 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2362 self
.record_event("config-failed", "NS configuration failed",
2363 evt_details
=self
._config
_status
_details
)
2365 yield from self
.publish()
2368 def is_active(self
):
2369 """ This NS is active """
2370 self
.set_state(NetworkServiceRecordState
.RUNNING
)
2374 # Publish the NSR to DTS
2375 self
._log
.debug("Network service %s is active ", self
.id)
2376 self
._is
_active
= True
2378 event_descr
= "NSR in running state for NSR id %s" % self
.id
2379 self
.record_event("ns-running", event_descr
)
2381 yield from self
.publish()
2384 def instantiation_failed(self
, failed_reason
=None):
2385 """ The NS instantiation failed"""
2386 self
._log
.error("Network service id:%s, name:%s instantiation failed",
2388 self
.set_state(NetworkServiceRecordState
.FAILED
)
2390 event_descr
= "Instantiation of NS %s failed" % self
.id
2391 self
.record_event("ns-failed", event_descr
, evt_details
=failed_reason
)
2393 # Publish the NSR to DTS
2394 yield from self
.publish()
2397 def terminate_vnfrs(self
, vnfrs
):
2398 """ Terminate VNFRS in this network service """
2399 self
._log
.debug("Terminating VNFs in network service %s", self
.id)
2401 yield from self
.nsm_plugin
.terminate_vnf(vnfr
)
2404 def terminate(self
):
2405 """ Terminate a NetworkServiceRecord."""
2406 def terminate_vnffgrs():
2407 """ Terminate VNFFGRS in this network service """
2408 self
._log
.debug("Terminating VNFFGRs in network service %s", self
.id)
2409 for vnffgr
in self
.vnffgrs
.values():
2410 yield from vnffgr
.terminate()
2412 def terminate_vlrs():
2413 """ Terminate VLRs in this netork service """
2414 self
._log
.debug("Terminating VLs in network service %s", self
.id)
2415 for vlr
in self
.vlrs
:
2416 yield from self
.nsm_plugin
.terminate_vl(vlr
)
2417 vlr
.state
= VlRecordState
.TERMINATED
2419 self
._log
.debug("Terminating network service id %s", self
.id)
2421 # Move the state to TERMINATE
2422 self
.set_state(NetworkServiceRecordState
.TERMINATE
)
2423 event_descr
= "Terminate being processed for NS Id:%s" % self
.id
2424 self
.record_event("terminate", event_descr
)
2426 # Move the state to VNF_TERMINATE_PHASE
2427 self
._log
.debug("Terminating VNFFGs in NS ID: %s", self
.id)
2428 self
.set_state(NetworkServiceRecordState
.VNFFG_TERMINATE_PHASE
)
2429 event_descr
= "Terminating VNFFGS in NS Id:%s" % self
.id
2430 self
.record_event("terminating-vnffgss", event_descr
)
2431 yield from terminate_vnffgrs()
2433 # Move the state to VNF_TERMINATE_PHASE
2434 self
.set_state(NetworkServiceRecordState
.VNF_TERMINATE_PHASE
)
2435 event_descr
= "Terminating VNFS in NS Id:%s" % self
.id
2436 self
.record_event("terminating-vnfs", event_descr
)
2437 yield from self
.terminate_vnfrs(self
.vnfrs
.values())
2439 # Move the state to VL_TERMINATE_PHASE
2440 self
.set_state(NetworkServiceRecordState
.VL_TERMINATE_PHASE
)
2441 event_descr
= "Terminating VLs in NS Id:%s" % self
.id
2442 self
.record_event("terminating-vls", event_descr
)
2443 yield from terminate_vlrs()
2445 yield from self
.nsm_plugin
.terminate_ns(self
)
2447 # Move the state to TERMINATED
2448 self
.set_state(NetworkServiceRecordState
.TERMINATED
)
2449 event_descr
= "Terminated NS Id:%s" % self
.id
2450 self
.record_event("terminated", event_descr
)
2453 """"Enable a NetworkServiceRecord."""
2457 """"Disable a NetworkServiceRecord."""
2460 def map_config_status(self
):
2461 self
._log
.debug("Config status for ns {} is {}".
2462 format(self
.name
, self
._config
_status
))
2463 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURING
:
2464 return 'configuring'
2465 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2469 def vl_phase_completed(self
):
2470 """ Are VLs created in this NS?"""
2471 return self
._vl
_phase
_completed
2473 def vnf_phase_completed(self
):
2474 """ Are VLs created in this NS?"""
2475 return self
._vnf
_phase
_completed
2477 def create_msg(self
):
2478 """ The network serice record as a message """
2479 nsr_dict
= {"ns_instance_config_ref": self
.id}
2480 nsr
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr
.from_dict(nsr_dict
)
2481 #nsr.cloud_account = self.cloud_account_name
2482 nsr
.sdn_account
= self
._sdn
_account
_name
2483 nsr
.name_ref
= self
.name
2484 nsr
.nsd_ref
= self
.nsd_id
2485 nsr
.nsd_name_ref
= self
.nsd_msg
.name
2486 nsr
.operational_events
= self
._op
_status
.msg
2487 nsr
.operational_status
= self
._op
_status
.yang_str()
2488 nsr
.config_status
= self
.map_config_status()
2489 nsr
.config_status_details
= self
._config
_status
_details
2490 nsr
.create_time
= self
._create
_time
2491 nsr
.uptime
= int(time
.time()) - self
._create
_time
2493 for cfg_prim
in self
.nsd_msg
.service_primitive
:
2494 cfg_prim
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_ServicePrimitive
.from_dict(
2496 nsr
.service_primitive
.append(cfg_prim
)
2498 for init_cfg
in self
.nsd_msg
.initial_config_primitive
:
2499 prim
= NsrYang
.NsrInitialConfigPrimitive
.from_dict(
2501 nsr
.initial_config_primitive
.append(prim
)
2503 if self
.vl_phase_completed():
2504 for vlr
in self
.vlrs
:
2505 nsr
.vlr
.append(vlr
.create_nsr_vlr_msg(self
.vnfrs
.values()))
2507 if self
.vnf_phase_completed():
2508 for vnfr_id
in self
.vnfrs
:
2509 nsr
.constituent_vnfr_ref
.append(self
.vnfrs
[vnfr_id
].const_vnfr_msg
)
2510 for vnffgr
in self
.vnffgrs
.values():
2511 nsr
.vnffgr
.append(vnffgr
.fetch_vnffgr())
2512 for scaling_group
in self
._scaling
_groups
.values():
2513 nsr
.scaling_group_record
.append(scaling_group
.create_record_msg())
2517 def all_vnfs_active(self
):
2518 """ Are all VNFS in this NS active? """
2519 for _
, vnfr
in self
.vnfrs
.items():
2520 if vnfr
.active
is not True:
2525 def update_state(self
):
2526 """ Re-evaluate this NS's state """
2527 curr_state
= self
._op
_status
.state
2529 if curr_state
== NetworkServiceRecordState
.TERMINATED
:
2530 self
._log
.debug("NS (%s) in terminated state, not updating state", self
.id)
2533 new_state
= NetworkServiceRecordState
.RUNNING
2534 self
._log
.info("Received update_state for nsr: %s, curr-state: %s",
2535 self
.id, curr_state
)
2537 # Check all the VNFRs are present
2538 for _
, vnfr
in self
.vnfrs
.items():
2539 if vnfr
.state
in [VnfRecordState
.ACTIVE
, VnfRecordState
.TERMINATED
]:
2541 elif vnfr
.state
== VnfRecordState
.FAILED
:
2542 if vnfr
._prev
_state
!= vnfr
.state
:
2543 event_descr
= "Instantiation of VNF %s failed" % vnfr
.id
2544 event_error_details
= vnfr
.state_failed_reason
2545 self
.record_event("vnf-failed", event_descr
, evt_details
=event_error_details
)
2546 vnfr
.set_state(VnfRecordState
.FAILED
)
2548 self
._log
.info("VNF state did not change, curr=%s, prev=%s",
2549 vnfr
.state
, vnfr
._prev
_state
)
2550 new_state
= NetworkServiceRecordState
.FAILED
2553 self
._log
.info("VNF %s in NSR %s is still not active; current state is: %s",
2554 vnfr
.id, self
.id, vnfr
.state
)
2555 new_state
= curr_state
2557 # If new state is RUNNING; check all VLs
2558 if new_state
== NetworkServiceRecordState
.RUNNING
:
2559 for vl
in self
.vlrs
:
2561 if vl
.state
in [VlRecordState
.ACTIVE
, VlRecordState
.TERMINATED
]:
2563 elif vl
.state
== VlRecordState
.FAILED
:
2564 if vl
.prev_state
!= vl
.state
:
2565 event_descr
= "Instantiation of VL %s failed" % vl
.id
2566 event_error_details
= vl
.state_failed_reason
2567 self
.record_event("vl-failed", event_descr
, evt_details
=event_error_details
)
2568 vl
.prev_state
= vl
.state
2570 self
._log
.debug("VL %s already in failed state")
2572 if vl
.state
in [VlRecordState
.INSTANTIATION_PENDING
, VlRecordState
.INIT
]:
2573 new_state
= NetworkServiceRecordState
.VL_INSTANTIATE
2576 if vl
.state
in [VlRecordState
.TERMINATE_PENDING
]:
2577 new_state
= NetworkServiceRecordState
.VL_TERMINATE
2580 # If new state is RUNNING; check VNFFGRs are also active
2581 if new_state
== NetworkServiceRecordState
.RUNNING
:
2582 for _
, vnffgr
in self
.vnffgrs
.items():
2583 self
._log
.info("Checking vnffgr state for nsr %s is: %s",
2584 self
.id, vnffgr
.state
)
2585 if vnffgr
.state
== VnffgRecordState
.ACTIVE
:
2587 elif vnffgr
.state
== VnffgRecordState
.FAILED
:
2588 event_descr
= "Instantiation of VNFFGR %s failed" % vnffgr
.id
2589 self
.record_event("vnffg-failed", event_descr
)
2590 new_state
= NetworkServiceRecordState
.FAILED
2593 self
._log
.info("VNFFGR %s in NSR %s is still not active; current state is: %s",
2594 vnffgr
.id, self
.id, vnffgr
.state
)
2595 new_state
= curr_state
2597 # Update all the scaling group instance operational status to
2598 # reflect the state of all VNFR within that instance
2599 yield from self
._update
_scale
_group
_instances
_status
()
2601 for _
, group
in self
._scaling
_groups
.items():
2602 if group
.state
== scale_group
.ScaleGroupState
.SCALING_OUT
:
2603 new_state
= NetworkServiceRecordState
.SCALING_OUT
2605 elif group
.state
== scale_group
.ScaleGroupState
.SCALING_IN
:
2606 new_state
= NetworkServiceRecordState
.SCALING_IN
2609 if new_state
!= curr_state
:
2610 self
._log
.debug("Changing state of Network service %s from %s to %s",
2611 self
.id, curr_state
, new_state
)
2612 if new_state
== NetworkServiceRecordState
.RUNNING
:
2613 yield from self
.is_active()
2614 elif new_state
== NetworkServiceRecordState
.FAILED
:
2615 # If the NS is already active and we entered scaling_in, scaling_out,
2616 # do not mark the NS as failing if scaling operation failed.
2617 if curr_state
in [NetworkServiceRecordState
.SCALING_OUT
,
2618 NetworkServiceRecordState
.SCALING_IN
] and self
._is
_active
:
2619 new_state
= NetworkServiceRecordState
.RUNNING
2620 self
.set_state(new_state
)
2622 yield from self
.instantiation_failed()
2624 self
.set_state(new_state
)
2626 yield from self
.publish()
2629 class InputParameterSubstitution(object):
2631 This class is responsible for substituting input parameters into an NSD.
2634 def __init__(self
, log
):
2635 """Create an instance of InputParameterSubstitution
2638 log - a logger for this object to use
2643 def __call__(self
, nsd
, nsr_config
):
2644 """Substitutes input parameters from the NSR config into the NSD
2646 This call modifies the provided NSD with the input parameters that are
2647 contained in the NSR config.
2650 nsd - a GI NSD object
2651 nsr_config - a GI NSR config object
2654 if nsd
is None or nsr_config
is None:
2657 # Create a lookup of the xpath elements that this descriptor allows
2659 optional_input_parameters
= set()
2660 for input_parameter
in nsd
.input_parameter_xpath
:
2661 optional_input_parameters
.add(input_parameter
.xpath
)
2663 # Apply the input parameters to the descriptor
2664 if nsr_config
.input_parameter
:
2665 for param
in nsr_config
.input_parameter
:
2666 if param
.xpath
not in optional_input_parameters
:
2667 msg
= "tried to set an invalid input parameter ({})"
2668 self
.log
.error(msg
.format(param
.xpath
))
2672 "input-parameter:{} = {}".format(
2679 xpath
.setxattr(nsd
, param
.xpath
, param
.value
)
2681 except Exception as e
:
2682 self
.log
.exception(e
)
2685 class NetworkServiceDescriptor(object):
2687 Network service descriptor class
2690 def __init__(self
, dts
, log
, loop
, nsd
, nsm
):
2702 """ Returns nsd id """
2707 """ Returns name of nsd """
2708 return self
._nsd
.name
2711 def ref_count(self
):
2712 """ Returns reference count"""
2713 return self
._ref
_count
2716 """ Returns whether nsd is in use or not """
2717 return True if self
.ref_count
> 0 else False
2720 """ Take a reference on this object """
2721 self
._ref
_count
+= 1
2724 """ Release reference on this object """
2725 if self
.ref_count
< 1:
2726 msg
= ("Unref on a NSD object - nsd id %s, ref_count = %s" %
2727 (self
.id, self
.ref_count
))
2728 self
._log
.critical(msg
)
2729 raise NetworkServiceDescriptorError(msg
)
2730 self
._ref
_count
-= 1
2734 """ Return the message associated with this NetworkServiceDescriptor"""
2738 def path_for_id(nsd_id
):
2739 """ Return path for the passed nsd_id"""
2740 return "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}'".format(nsd_id
)
2743 """ Return the message associated with this NetworkServiceDescriptor"""
2744 return NetworkServiceDescriptor
.path_for_id(self
.id)
2746 def update(self
, nsd
):
2747 """ Update the NSD descriptor """
2751 class NsdDtsHandler(object):
2752 """ The network service descriptor DTS handler """
2753 XPATH
= "C,/nsd:nsd-catalog/nsd:nsd"
2755 def __init__(self
, dts
, log
, loop
, nsm
):
2765 """ Return registration handle """
2770 """ Register for Nsd create/update/delete/read requests from dts """
2772 def on_apply(dts
, acg
, xact
, action
, scratch
):
2773 """Apply the configuration"""
2774 is_recovery
= xact
.xact
is None and action
== rwdts
.AppconfAction
.INSTALL
2775 self
._log
.debug("Got nsd apply cfg (xact:%s) (action:%s)",
2777 # Create/Update an NSD record
2778 for cfg
in self
._regh
.get_xact_elements(xact
):
2779 # Only interested in those NSD cfgs whose ID was received in prepare callback
2780 if cfg
.id in scratch
.get('nsds', []) or is_recovery
:
2781 self
._nsm
.update_nsd(cfg
)
2783 scratch
.pop('nsds', None)
2785 return RwTypes
.RwStatus
.SUCCESS
2788 def delete_nsd_libs(nsd_id
):
2789 """ Remove any files uploaded with NSD and stored under $RIFT_ARTIFACTS/libs/<id> """
2791 rift_artifacts_dir
= os
.environ
['RIFT_ARTIFACTS']
2792 nsd_dir
= os
.path
.join(rift_artifacts_dir
, 'launchpad/libs', nsd_id
)
2794 if os
.path
.exists (nsd_dir
):
2795 shutil
.rmtree(nsd_dir
, ignore_errors
=True)
2796 except Exception as e
:
2797 self
._log
.error("Exception in cleaning up NSD libs {}: {}".
2799 self
._log
.excpetion(e
)
2802 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2803 """ Prepare callback from DTS for NSD config """
2805 self
._log
.info("Got nsd prepare - config received nsd id %s, msg %s",
2808 fref
= ProtobufC
.FieldReference
.alloc()
2809 fref
.goto_whole_message(msg
.to_pbcm())
2811 if fref
.is_field_deleted():
2812 # Delete an NSD record
2813 self
._log
.debug("Deleting NSD with id %s", msg
.id)
2814 if self
._nsm
.nsd_in_use(msg
.id):
2815 self
._log
.debug("Cannot delete NSD in use - %s", msg
.id)
2816 err
= "Cannot delete an NSD in use - %s" % msg
.id
2817 raise NetworkServiceDescriptorRefCountExists(err
)
2819 yield from delete_nsd_libs(msg
.id)
2820 self
._nsm
.delete_nsd(msg
.id)
2822 # Add this NSD to scratch to create/update in apply callback
2823 nsds
= scratch
.setdefault('nsds', [])
2825 # acg._scratch['nsds'].append(msg.id)
2827 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2830 "Registering for NSD config using xpath: %s",
2831 NsdDtsHandler
.XPATH
,
2834 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2835 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2836 # Need a list in scratch to store NSDs to create/update later
2837 # acg._scratch['nsds'] = list()
2838 self
._regh
= acg
.register(
2839 xpath
=NsdDtsHandler
.XPATH
,
2840 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
2841 on_prepare
=on_prepare
)
2844 class VnfdDtsHandler(object):
2845 """ DTS handler for VNFD config changes """
2846 XPATH
= "C,/vnfd:vnfd-catalog/vnfd:vnfd"
2848 def __init__(self
, dts
, log
, loop
, nsm
):
2857 """ DTS registration handle """
2862 """ Register for VNFD configuration"""
2865 def on_apply(dts
, acg
, xact
, action
, scratch
):
2866 """Apply the configuration"""
2867 self
._log
.debug("Got NSM VNFD apply (xact: %s) (action: %s)(scr: %s)",
2868 xact
, action
, scratch
)
2870 # Create/Update a VNFD record
2871 for cfg
in self
._regh
.get_xact_elements(xact
):
2872 # Only interested in those VNFD cfgs whose ID was received in prepare callback
2873 if cfg
.id in scratch
.get('vnfds', []):
2874 self
._nsm
.update_vnfd(cfg
)
2876 for cfg
in self
._regh
.elements
:
2877 if cfg
.id in scratch
.get('deleted_vnfds', []):
2878 yield from self
._nsm
.delete_vnfd(cfg
.id)
2880 scratch
.pop('vnfds', None)
2881 scratch
.pop('deleted_vnfds', None)
2884 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2885 """ on prepare callback """
2886 self
._log
.debug("Got on prepare for VNFD (path: %s) (action: %s) (msg: %s)",
2887 ks_path
.to_xpath(RwNsmYang
.get_schema()), xact_info
.query_action
, msg
)
2889 fref
= ProtobufC
.FieldReference
.alloc()
2890 fref
.goto_whole_message(msg
.to_pbcm())
2892 # Handle deletes in prepare_callback, but adds/updates in apply_callback
2893 if fref
.is_field_deleted():
2894 self
._log
.debug("Adding msg to deleted field")
2895 deleted_vnfds
= scratch
.setdefault('deleted_vnfds', [])
2896 deleted_vnfds
.append(msg
.id)
2898 # Add this VNFD to scratch to create/update in apply callback
2899 vnfds
= scratch
.setdefault('vnfds', [])
2900 vnfds
.append(msg
.id)
2902 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2905 "Registering for VNFD config using xpath: %s",
2906 VnfdDtsHandler
.XPATH
,
2908 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2909 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2910 # Need a list in scratch to store VNFDs to create/update later
2911 # acg._scratch['vnfds'] = list()
2912 # acg._scratch['deleted_vnfds'] = list()
2913 self
._regh
= acg
.register(
2914 xpath
=VnfdDtsHandler
.XPATH
,
2915 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY
,
2916 on_prepare
=on_prepare
)
2918 class NsrRpcDtsHandler(object):
2919 """ The network service instantiation RPC DTS handler """
2920 EXEC_NSR_CONF_XPATH
= "I,/nsr:start-network-service"
2921 EXEC_NSR_CONF_O_XPATH
= "O,/nsr:start-network-service"
2922 NETCONF_IP_ADDRESS
= "127.0.0.1"
2924 RESTCONF_PORT
= 8888
2925 NETCONF_USER
= "admin"
2926 NETCONF_PW
= "admin"
2927 REST_BASE_V2_URL
= 'https://{}:{}/v2/api/'.format("127.0.0.1",8888)
2929 def __init__(self
, dts
, log
, loop
, nsm
):
2936 self
._ns
_regh
= None
2938 self
._manager
= None
2939 self
._nsr
_config
_url
= NsrRpcDtsHandler
.REST_BASE_V2_URL
+ 'config/ns-instance-config'
2941 self
._model
= RwYang
.Model
.create_libncx()
2942 self
._model
.load_schema_ypbc(RwNsrYang
.get_schema())
2946 """ Return the NS manager instance """
2950 def wrap_netconf_config_xml(xml
):
2951 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
2955 def _connect(self
, timeout_secs
=240):
2957 start_time
= time
.time()
2958 while (time
.time() - start_time
) < timeout_secs
:
2961 self
._log
.debug("Attemping NsmTasklet netconf connection.")
2963 manager
= yield from ncclient
.asyncio_manager
.asyncio_connect(
2965 host
=NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
,
2966 port
=NsrRpcDtsHandler
.NETCONF_PORT
,
2967 username
=NsrRpcDtsHandler
.NETCONF_USER
,
2968 password
=NsrRpcDtsHandler
.NETCONF_PW
,
2970 look_for_keys
=False,
2971 hostkey_verify
=False,
2976 except ncclient
.transport
.errors
.SSHError
as e
:
2977 self
._log
.warning("Netconf connection to launchpad %s failed: %s",
2978 NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
, str(e
))
2980 yield from asyncio
.sleep(5, loop
=self
._loop
)
2982 raise NsrInstantiationFailed("Failed to connect to Launchpad within %s seconds" %
2985 def _apply_ns_instance_config(self
,payload_dict
):
2986 #self._log.debug("At apply NS instance config with payload %s",payload_dict)
2987 req_hdr
= {'accept':'application/vnd.yang.data+json','content-type':'application/vnd.yang.data+json'}
2988 response
=requests
.post(self
._nsr
_config
_url
, headers
=req_hdr
, auth
=('admin', 'admin'),data
=payload_dict
,verify
=False)
2993 """ Register for NS monitoring read from dts """
2995 def on_ns_config_prepare(xact_info
, action
, ks_path
, msg
):
2996 """ prepare callback from dts start-network-service"""
2997 assert action
== rwdts
.QueryAction
.RPC
2999 rpc_op
= NsrYang
.YangOutput_Nsr_StartNetworkService
.from_dict({
3000 "nsr_id":str(uuid
.uuid4())
3003 if not ('name' in rpc_ip
and 'nsd_ref' in rpc_ip
and ('cloud_account' in rpc_ip
or 'om_datacenter' in rpc_ip
)):
3004 self
._log
.error("Mandatory parameters name or nsd_ref or cloud account not found in start-network-service {}".format(rpc_ip
))
3007 self
._log
.debug("start-network-service RPC input: {}".format(rpc_ip
))
3010 # Add used value to the pool
3011 self
._log
.debug("RPC output: {}".format(rpc_op
))
3013 nsd_copy
= self
.nsm
.get_nsd(rpc_ip
.nsd_ref
)
3015 #if not self._manager:
3016 # self._manager = yield from self._connect()
3018 self
._log
.debug("Configuring ns-instance-config with name %s nsd-ref: %s",
3019 rpc_ip
.name
, rpc_ip
.nsd_ref
)
3021 ns_instance_config_dict
= {"id":rpc_op
.nsr_id
, "admin_status":"ENABLED"}
3022 ns_instance_config_copy_dict
= {k
:v
for k
, v
in rpc_ip
.as_dict().items()
3023 if k
in RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr().fields
}
3024 ns_instance_config_dict
.update(ns_instance_config_copy_dict
)
3026 ns_instance_config
= RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.from_dict(ns_instance_config_dict
)
3027 ns_instance_config
.nsd
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_Nsd()
3028 ns_instance_config
.nsd
.from_dict(nsd_copy
.msg
.as_dict())
3030 payload_dict
= ns_instance_config
.to_json(self
._model
)
3031 #xml = ns_instance_config.to_xml_v2(self._model)
3032 #netconf_xml = self.wrap_netconf_config_xml(xml)
3034 #self._log.debug("Sending configure ns-instance-config xml to %s: %s",
3035 # netconf_xml, NsrRpcDtsHandler.NETCONF_IP_ADDRESS)
3036 self
._log
.debug("Sending configure ns-instance-config json to %s: %s",
3037 self
._nsr
_config
_url
,ns_instance_config
)
3039 #response = yield from self._manager.edit_config(
3041 # config=netconf_xml,
3043 response
= yield from self
._loop
.run_in_executor(
3045 self
._apply
_ns
_instance
_config
,
3048 response
.raise_for_status()
3049 self
._log
.debug("Received edit config response: %s", response
.json())
3051 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
,
3052 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
,
3054 except Exception as e
:
3055 self
._log
.error("Exception processing the "
3056 "start-network-service: {}".format(e
))
3057 self
._log
.exception(e
)
3058 xact_info
.respond_xpath(rwdts
.XactRspCode
.NACK
,
3059 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
)
3062 hdl_ns
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_ns_config_prepare
,)
3064 with self
._dts
.group_create() as group
:
3065 self
._ns
_regh
= group
.register(xpath
=NsrRpcDtsHandler
.EXEC_NSR_CONF_XPATH
,
3067 flags
=rwdts
.Flag
.PUBLISHER
,
3071 class NsrDtsHandler(object):
3072 """ The network service DTS handler """
3073 NSR_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr"
3074 SCALE_INSTANCE_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr/nsr:scaling-group/nsr:instance"
3075 KEY_PAIR_XPATH
= "C,/nsr:key-pair"
3077 def __init__(self
, dts
, log
, loop
, nsm
):
3083 self
._nsr
_regh
= None
3084 self
._scale
_regh
= None
3085 self
._key
_pair
_regh
= None
3089 """ Return the NS manager instance """
3094 """ Register for Nsr create/update/delete/read requests from dts """
3096 def nsr_id_from_keyspec(ks
):
3097 nsr_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.schema().keyspec_to_entry(ks
)
3098 nsr_id
= nsr_path_entry
.key00
.id
3101 def group_name_from_keyspec(ks
):
3102 group_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
.schema().keyspec_to_entry(ks
)
3103 group_name
= group_path_entry
.key00
.scaling_group_name_ref
3106 def is_instance_in_reg_elements(nsr_id
, group_name
, instance_id
):
3107 """ Return boolean indicating if scaling group instance was already commited previously.
3109 By looking at the existing elements in this registration handle (elements not part
3110 of this current xact), we can tell if the instance was configured previously without
3111 keeping any application state.
3113 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3114 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3115 elem_group_name
= group_name_from_keyspec(keyspec
)
3117 if elem_nsr_id
!= nsr_id
or group_name
!= elem_group_name
:
3120 if instance_cfg
.id == instance_id
:
3125 def get_scale_group_instance_delta(nsr_id
, group_name
, xact
):
3126 delta
= {"added": [], "deleted": []}
3127 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3128 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3129 if elem_nsr_id
!= nsr_id
:
3132 elem_group_name
= group_name_from_keyspec(keyspec
)
3133 if elem_group_name
!= group_name
:
3136 delta
["added"].append(instance_cfg
.id)
3138 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(include_keyspec
=True):
3139 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3140 if elem_nsr_id
!= nsr_id
:
3143 elem_group_name
= group_name_from_keyspec(keyspec
)
3144 if elem_group_name
!= group_name
:
3147 if instance_cfg
.id in delta
["added"]:
3148 delta
["added"].remove(instance_cfg
.id)
3150 delta
["deleted"].append(instance_cfg
.id)
3155 def update_nsr_nsd(nsr_id
, xact
, scratch
):
3158 def get_nsr_vl_delta(nsr_id
, xact
, scratch
):
3159 delta
= {"added": [], "deleted": []}
3160 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3161 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3162 if elem_nsr_id
!= nsr_id
:
3165 if 'vld' in instance_cfg
.nsd
:
3166 for vld
in instance_cfg
.nsd
.vld
:
3167 delta
["added"].append(vld
)
3169 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3170 self
._log
.debug("NSR update: %s", instance_cfg
)
3171 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3172 if elem_nsr_id
!= nsr_id
:
3175 if 'vld' in instance_cfg
.nsd
:
3176 for vld
in instance_cfg
.nsd
.vld
:
3177 if vld
in delta
["added"]:
3178 delta
["added"].remove(vld
)
3180 delta
["deleted"].append(vld
)
3184 vl_delta
= yield from get_nsr_vl_delta(nsr_id
, xact
, scratch
)
3185 self
._log
.debug("Got NSR:%s VL instance delta: %s", nsr_id
, vl_delta
)
3187 for vld
in vl_delta
["added"]:
3188 yield from self
._nsm
.nsr_instantiate_vl(nsr_id
, vld
)
3190 for vld
in vl_delta
["deleted"]:
3191 yield from self
._nsm
.nsr_terminate_vl(nsr_id
, vld
)
3193 def get_add_delete_update_cfgs(dts_member_reg
, xact
, key_name
, scratch
):
3194 # Unfortunately, it is currently difficult to figure out what has exactly
3195 # changed in this xact without Pbdelta support (RIFT-4916)
3196 # As a workaround, we can fetch the pre and post xact elements and
3197 # perform a comparison to figure out adds/deletes/updates
3198 xact_cfgs
= list(dts_member_reg
.get_xact_elements(xact
))
3199 curr_cfgs
= list(dts_member_reg
.elements
)
3201 xact_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in xact_cfgs
}
3202 curr_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in curr_cfgs
}
3205 added_keys
= set(xact_key_map
) - set(curr_key_map
)
3206 added_cfgs
= [xact_key_map
[key
] for key
in added_keys
]
3209 deleted_keys
= set(curr_key_map
) - set(xact_key_map
)
3210 deleted_cfgs
= [curr_key_map
[key
] for key
in deleted_keys
]
3213 updated_keys
= set(curr_key_map
) & set(xact_key_map
)
3214 updated_cfgs
= [xact_key_map
[key
] for key
in updated_keys
3215 if xact_key_map
[key
] != curr_key_map
[key
]]
3217 return added_cfgs
, deleted_cfgs
, updated_cfgs
3219 def get_nsr_key_pairs(dts_member_reg
, xact
):
3221 for instance_cfg
, keyspec
in dts_member_reg
.get_xact_elements(xact
, include_keyspec
=True):
3222 self
._log
.debug("Key pair received is {} KS: {}".format(instance_cfg
, keyspec
))
3223 xpath
= keyspec
.to_xpath(RwNsrYang
.get_schema())
3224 key_pairs
[instance_cfg
.name
] = instance_cfg
3227 def on_apply(dts
, acg
, xact
, action
, scratch
):
3228 """Apply the configuration"""
3229 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3230 xact
, action
, scratch
)
3232 def handle_create_nsr(msg
, key_pairs
=None, restart_mode
=False):
3233 # Handle create nsr requests """
3234 # Do some validations
3235 if not msg
.has_field("nsd"):
3236 err
= "NSD not provided"
3237 self
._log
.error(err
)
3238 raise NetworkServiceRecordError(err
)
3240 self
._log
.debug("Creating NetworkServiceRecord %s from nsr config %s",
3241 msg
.id, msg
.as_dict())
3242 nsr
= self
.nsm
.create_nsr(msg
, key_pairs
=key_pairs
, restart_mode
=restart_mode
)
3245 def handle_delete_nsr(msg
):
3247 def delete_instantiation(ns_id
):
3248 """ Delete instantiation """
3249 with self
._dts
.transaction() as xact
:
3250 yield from self
._nsm
.terminate_ns(ns_id
, xact
)
3252 # Handle delete NSR requests
3253 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3254 # Terminate the NSR instance
3255 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3257 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3258 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3259 nsr
.record_event("terminate-rcvd", event_descr
)
3261 self
._loop
.create_task(delete_instantiation(msg
.id))
3264 def begin_instantiation(nsr
):
3265 # Begin instantiation
3266 self
._log
.info("Beginning NS instantiation: %s", nsr
.id)
3267 yield from self
._nsm
.instantiate_ns(nsr
.id, xact
)
3269 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3270 xact
, action
, scratch
)
3272 if action
== rwdts
.AppconfAction
.INSTALL
and xact
.id is None:
3274 for element
in self
._key
_pair
_regh
.elements
:
3275 key_pairs
.append(element
)
3276 for element
in self
._nsr
_regh
.elements
:
3277 nsr
= handle_create_nsr(element
, key_pairs
, restart_mode
=True)
3278 self
._loop
.create_task(begin_instantiation(nsr
))
3281 (added_msgs
, deleted_msgs
, updated_msgs
) = get_add_delete_update_cfgs(self
._nsr
_regh
,
3285 self
._log
.debug("Added: %s, Deleted: %s, Updated: %s", added_msgs
,
3286 deleted_msgs
, updated_msgs
)
3288 for msg
in added_msgs
:
3289 if msg
.id not in self
._nsm
.nsrs
:
3290 self
._log
.info("Create NSR received in on_apply to instantiate NS:%s", msg
.id)
3291 key_pairs
= get_nsr_key_pairs(self
._key
_pair
_regh
, xact
)
3292 nsr
= handle_create_nsr(msg
,key_pairs
)
3293 self
._loop
.create_task(begin_instantiation(nsr
))
3295 for msg
in deleted_msgs
:
3296 self
._log
.info("Delete NSR received in on_apply to terminate NS:%s", msg
.id)
3298 handle_delete_nsr(msg
)
3300 self
._log
.exception("Failed to terminate NS:%s", msg
.id)
3302 for msg
in updated_msgs
:
3303 self
._log
.info("Update NSR received in on_apply: %s", msg
)
3305 self
._nsm
.nsr_update_cfg(msg
.id, msg
)
3308 self
._loop
.create_task(update_nsr_nsd(msg
.id, xact
, scratch
))
3310 for group
in msg
.scaling_group
:
3311 instance_delta
= get_scale_group_instance_delta(msg
.id, group
.scaling_group_name_ref
, xact
)
3312 self
._log
.debug("Got NSR:%s scale group instance delta: %s", msg
.id, instance_delta
)
3314 for instance_id
in instance_delta
["added"]:
3315 self
._nsm
.scale_nsr_out(msg
.id, group
.scaling_group_name_ref
, instance_id
, xact
)
3317 for instance_id
in instance_delta
["deleted"]:
3318 self
._nsm
.scale_nsr_in(msg
.id, group
.scaling_group_name_ref
, instance_id
)
3321 return RwTypes
.RwStatus
.SUCCESS
3324 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
3325 """ Prepare calllback from DTS for NSR """
3327 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3328 action
= xact_info
.query_action
3330 "Got Nsr prepare callback (xact: %s) (action: %s) (info: %s), %s:%s)",
3331 xact
, action
, xact_info
, xpath
, msg
3335 def delete_instantiation(ns_id
):
3336 """ Delete instantiation """
3337 yield from self
._nsm
.terminate_ns(ns_id
, None)
3339 def handle_delete_nsr():
3340 """ Handle delete NSR requests """
3341 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3342 # Terminate the NSR instance
3343 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3345 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3346 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3347 nsr
.record_event("terminate-rcvd", event_descr
)
3349 self
._loop
.create_task(delete_instantiation(msg
.id))
3351 fref
= ProtobufC
.FieldReference
.alloc()
3352 fref
.goto_whole_message(msg
.to_pbcm())
3354 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
, rwdts
.QueryAction
.DELETE
]:
3355 # if this is an NSR create
3356 if action
!= rwdts
.QueryAction
.DELETE
and msg
.id not in self
._nsm
.nsrs
:
3357 # Ensure the Cloud account/datacenter has been specified
3358 if not msg
.has_field("cloud_account") and not msg
.has_field("om_datacenter"):
3359 raise NsrInstantiationFailed("Cloud account or datacenter not specified in NSR")
3361 # Check if nsd is specified
3362 if not msg
.has_field("nsd"):
3363 raise NsrInstantiationFailed("NSD not specified in NSR")
3366 nsr
= self
._nsm
.nsrs
[msg
.id]
3368 if msg
.has_field("nsd"):
3369 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3370 raise NsrVlUpdateError("Unable to update VL when NSR not in running state")
3371 if 'vld' not in msg
.nsd
or len(msg
.nsd
.vld
) == 0:
3372 raise NsrVlUpdateError("NS config NSD should have atleast 1 VLD defined")
3374 if msg
.has_field("scaling_group"):
3375 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3376 raise ScalingOperationError("Unable to perform scaling action when NS is not in running state")
3378 if len(msg
.scaling_group
) > 1:
3379 raise ScalingOperationError("Only a single scaling group can be configured at a time")
3381 for group_msg
in msg
.scaling_group
:
3382 num_new_group_instances
= len(group_msg
.instance
)
3383 if num_new_group_instances
> 1:
3384 raise ScalingOperationError("Only a single scaling instance can be modified at a time")
3386 elif num_new_group_instances
== 1:
3387 scale_group
= nsr
.scaling_groups
[group_msg
.scaling_group_name_ref
]
3388 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
]:
3389 if len(scale_group
.instances
) == scale_group
.max_instance_count
:
3390 raise ScalingOperationError("Max instances for %s reached" % scale_group
)
3392 acg
.handle
.prepare_complete_ok(xact_info
.handle
)
3395 self
._log
.debug("Registering for NSR config using xpath: %s",
3396 NsrDtsHandler
.NSR_XPATH
)
3398 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
3399 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
3400 self
._nsr
_regh
= acg
.register(xpath
=NsrDtsHandler
.NSR_XPATH
,
3401 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3402 on_prepare
=on_prepare
)
3404 self
._scale
_regh
= acg
.register(
3405 xpath
=NsrDtsHandler
.SCALE_INSTANCE_XPATH
,
3406 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY| rwdts
.Flag
.CACHE
,
3409 self
._key
_pair
_regh
= acg
.register(
3410 xpath
=NsrDtsHandler
.KEY_PAIR_XPATH
,
3411 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3415 class NsrOpDataDtsHandler(object):
3416 """ The network service op data DTS handler """
3417 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
3419 def __init__(self
, dts
, log
, loop
, nsm
):
3428 """ Return the registration handle"""
3433 """ Return the NS manager instance """
3438 """ Register for Nsr op data publisher registration"""
3439 self
._log
.debug("Registering Nsr op data path %s as publisher",
3440 NsrOpDataDtsHandler
.XPATH
)
3442 hdl
= rift
.tasklets
.DTS
.RegistrationHandler()
3443 handlers
= rift
.tasklets
.Group
.Handler()
3444 with self
._dts
.group_create(handler
=handlers
) as group
:
3445 self
._regh
= group
.register(xpath
=NsrOpDataDtsHandler
.XPATH
,
3447 flags
=rwdts
.Flag
.PUBLISHER | rwdts
.Flag
.NO_PREP_READ | rwdts
.Flag
.DATASTORE
)
3450 def create(self
, path
, msg
):
3452 Create an NS record in DTS with the path and message
3454 self
._log
.debug("Creating NSR %s:%s", path
, msg
)
3455 self
.regh
.create_element(path
, msg
)
3456 self
._log
.debug("Created NSR, %s:%s", path
, msg
)
3459 def update(self
, path
, msg
, flags
=rwdts
.XactFlag
.REPLACE
):
3461 Update an NS record in DTS with the path and message
3463 self
._log
.debug("Updating NSR, %s:%s regh = %s", path
, msg
, self
.regh
)
3464 self
.regh
.update_element(path
, msg
, flags
)
3465 self
._log
.debug("Updated NSR, %s:%s", path
, msg
)
3468 def delete(self
, path
):
3470 Update an NS record in DTS with the path and message
3472 self
._log
.debug("Deleting NSR path:%s", path
)
3473 self
.regh
.delete_element(path
)
3474 self
._log
.debug("Deleted NSR path:%s", path
)
3477 class VnfrDtsHandler(object):
3478 """ The virtual network service DTS handler """
3479 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
3481 def __init__(self
, dts
, log
, loop
, nsm
):
3491 """ Return registration handle """
3496 """ Return the NS manager instance """
3501 """ Register for vnfr create/update/delete/ advises from dts """
3503 def on_commit(xact_info
):
3504 """ The transaction has been committed """
3505 self
._log
.debug("Got vnfr commit (xact_info: %s)", xact_info
)
3506 return rwdts
.MemberRspCode
.ACTION_OK
3509 def on_prepare(xact_info
, action
, ks_path
, msg
):
3510 """ prepare callback from dts """
3511 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3513 "Got vnfr on_prepare cb (xact_info: %s, action: %s): %s:%s",
3514 xact_info
, action
, ks_path
, msg
3517 schema
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr
.schema()
3518 path_entry
= schema
.keyspec_to_entry(ks_path
)
3519 if path_entry
.key00
.id not in self
._nsm
._vnfrs
:
3520 self
._log
.error("%s request for non existent record path %s",
3522 xact_info
.respond_xpath(rwdts
.XactRspCode
.NA
, xpath
)
3526 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3527 if action
== rwdts
.QueryAction
.CREATE
or action
== rwdts
.QueryAction
.UPDATE
:
3528 yield from self
._nsm
.update_vnfr(msg
)
3529 elif action
== rwdts
.QueryAction
.DELETE
:
3530 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3531 self
._nsm
.delete_vnfr(path_entry
.key00
.id)
3533 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
, xpath
)
3535 self
._log
.debug("Registering for VNFR using xpath: %s",
3536 VnfrDtsHandler
.XPATH
,)
3538 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_commit
=on_commit
,
3539 on_prepare
=on_prepare
,)
3540 with self
._dts
.group_create() as group
:
3541 self
._regh
= group
.register(xpath
=VnfrDtsHandler
.XPATH
,
3543 flags
=(rwdts
.Flag
.SUBSCRIBER
),)
3546 class NsdRefCountDtsHandler(object):
3547 """ The NSD Ref Count DTS handler """
3548 XPATH
= "D,/nsr:ns-instance-opdata/rw-nsr:nsd-ref-count"
3550 def __init__(self
, dts
, log
, loop
, nsm
):
3560 """ Return registration handle """
3565 """ Return the NS manager instance """
3570 """ Register for NSD ref count read from dts """
3573 def on_prepare(xact_info
, action
, ks_path
, msg
):
3574 """ prepare callback from dts """
3575 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3577 if action
== rwdts
.QueryAction
.READ
:
3578 schema
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount
.schema()
3579 path_entry
= schema
.keyspec_to_entry(ks_path
)
3580 nsd_list
= yield from self
._nsm
.get_nsd_refcount(path_entry
.key00
.nsd_id_ref
)
3581 for xpath
, msg
in nsd_list
:
3582 xact_info
.respond_xpath(rsp_code
=rwdts
.XactRspCode
.MORE
,
3585 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
3587 raise NetworkServiceRecordError("Not supported operation %s" % action
)
3589 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_prepare
,)
3590 with self
._dts
.group_create() as group
:
3591 self
._regh
= group
.register(xpath
=NsdRefCountDtsHandler
.XPATH
,
3593 flags
=rwdts
.Flag
.PUBLISHER
,)
3596 class NsManager(object):
3597 """ The Network Service Manager class"""
3598 def __init__(self
, dts
, log
, loop
,
3599 nsr_handler
, vnfr_handler
, vlr_handler
, ro_plugin_selector
,
3600 vnffgmgr
, vnfd_pub_handler
, cloud_account_handler
):
3604 self
._nsr
_handler
= nsr_handler
3605 self
._vnfr
_pub
_handler
= vnfr_handler
3606 self
._vlr
_pub
_handler
= vlr_handler
3607 self
._vnffgmgr
= vnffgmgr
3608 self
._vnfd
_pub
_handler
= vnfd_pub_handler
3609 self
._cloud
_account
_handler
= cloud_account_handler
3611 self
._ro
_plugin
_selector
= ro_plugin_selector
3612 self
._ncclient
= rift
.mano
.ncclient
.NcClient(
3624 self
.cfgmgr_obj
= conman
.ROConfigManager(log
, loop
, dts
, self
)
3626 # TODO: All these handlers should move to tasklet level.
3627 # Passing self is often an indication of bad design
3628 self
._nsd
_dts
_handler
= NsdDtsHandler(dts
, log
, loop
, self
)
3629 self
._vnfd
_dts
_handler
= VnfdDtsHandler(dts
, log
, loop
, self
)
3630 self
._dts
_handlers
= [self
._nsd
_dts
_handler
,
3631 VnfrDtsHandler(dts
, log
, loop
, self
),
3632 NsdRefCountDtsHandler(dts
, log
, loop
, self
),
3633 NsrDtsHandler(dts
, log
, loop
, self
),
3634 ScalingRpcHandler(log
, dts
, loop
, self
.scale_rpc_callback
),
3635 NsrRpcDtsHandler(dts
,log
,loop
,self
),
3636 self
._vnfd
_dts
_handler
,
3657 def nsr_handler(self
):
3658 """" NSR handler """
3659 return self
._nsr
_handler
3663 """" So Obj handler """
3668 """ NSRs in this NSM"""
3673 """ NSDs in this NSM"""
3678 """ VNFDs in this NSM"""
3683 """ VNFRs in this NSM"""
3687 def nsr_pub_handler(self
):
3688 """ NSR publication handler """
3689 return self
._nsr
_handler
3692 def vnfr_pub_handler(self
):
3693 """ VNFR publication handler """
3694 return self
._vnfr
_pub
_handler
3697 def vlr_pub_handler(self
):
3698 """ VLR publication handler """
3699 return self
._vlr
_pub
_handler
3702 def vnfd_pub_handler(self
):
3703 return self
._vnfd
_pub
_handler
3707 """ Register all static DTS handlers """
3708 for dts_handle
in self
._dts
_handlers
:
3709 yield from dts_handle
.register()
3712 def get_ns_by_nsr_id(self
, nsr_id
):
3713 """ get NSR by nsr id """
3714 if nsr_id
not in self
._nsrs
:
3715 raise NetworkServiceRecordError("NSR id %s not found" % nsr_id
)
3717 return self
._nsrs
[nsr_id
]
3719 def scale_nsr_out(self
, nsr_id
, scale_group_name
, instance_id
, config_xact
):
3720 self
.log
.debug("Scale out NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3725 nsr
= self
._nsrs
[nsr_id
]
3726 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3727 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3729 self
._loop
.create_task(nsr
.create_scale_group_instance(scale_group_name
, instance_id
, config_xact
))
3731 def scale_nsr_in(self
, nsr_id
, scale_group_name
, instance_id
):
3732 self
.log
.debug("Scale in NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3737 nsr
= self
._nsrs
[nsr_id
]
3738 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3739 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3741 self
._loop
.create_task(nsr
.delete_scale_group_instance(scale_group_name
, instance_id
))
3743 def scale_rpc_callback(self
, xact
, msg
, action
):
3744 """Callback handler for RPC calls
3746 xact : Transaction Handler
3748 action : Scaling Action
3750 ScalingGroupInstance
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup_Instance
3751 ScalingGroup
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
3753 xpath
= ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').format(
3755 instance
= ScalingGroupInstance
.from_dict({"id": msg
.instance_id
})
3758 def get_nsr_scaling_group():
3759 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3761 for result
in results
:
3762 res
= yield from result
3763 nsr_config
= res
.result
3765 for scaling_group
in nsr_config
.scaling_group
:
3766 if scaling_group
.scaling_group_name_ref
== msg
.scaling_group_name_ref
:
3769 scaling_group
= nsr_config
.scaling_group
.add()
3770 scaling_group
.scaling_group_name_ref
= msg
.scaling_group_name_ref
3772 return (nsr_config
, scaling_group
)
3775 def update_config(nsr_config
):
3776 xml
= self
._ncclient
.convert_to_xml(RwNsrYang
, nsr_config
)
3777 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
3778 yield from self
._ncclient
.connect()
3779 yield from self
._ncclient
.manager
.edit_config(target
="running", config
=xml
, default_operation
="replace")
3783 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3784 scaling_group
.instance
.append(instance
)
3785 yield from update_config(nsr_config
)
3789 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3790 scaling_group
.instance
.remove(instance
)
3791 yield from update_config(nsr_config
)
3793 if action
== ScalingRpcHandler
.ACTION
.SCALE_OUT
:
3794 self
._loop
.create_task(scale_out())
3796 self
._loop
.create_task(scale_in())
3798 # Opdata based calls, disabled for now!
3799 # if action == ScalingRpcHandler.ACTION.SCALE_OUT:
3800 # self.scale_nsr_out(
3802 # msg.scaling_group_name_ref,
3806 # self.scale_nsr_in(
3808 # msg.scaling_group_name_ref,
3811 def nsr_update_cfg(self
, nsr_id
, msg
):
3812 nsr
= self
._nsrs
[nsr_id
]
3813 nsr
.nsr_cfg_msg
= msg
3815 def nsr_instantiate_vl(self
, nsr_id
, vld
):
3816 self
.log
.debug("NSR {} create VL {}".format(nsr_id
, vld
))
3817 nsr
= self
._nsrs
[nsr_id
]
3818 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3819 raise NsrVlUpdateError("Cannot perform VL instantiate if NSR is not in running state")
3821 # Not calling in a separate task as this is called from a separate task
3822 yield from nsr
.create_vl_instance(vld
)
3824 def nsr_terminate_vl(self
, nsr_id
, vld
):
3825 self
.log
.debug("NSR {} delete VL {}".format(nsr_id
, vld
.id))
3826 nsr
= self
._nsrs
[nsr_id
]
3827 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3828 raise NsrVlUpdateError("Cannot perform VL terminate 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
.delete_vl_instance(vld
)
3833 def create_nsr(self
, nsr_msg
, key_pairs
=None,restart_mode
=False):
3834 """ Create an NSR instance """
3835 if nsr_msg
.id in self
._nsrs
:
3836 msg
= "NSR id %s already exists" % nsr_msg
.id
3837 self
._log
.error(msg
)
3838 raise NetworkServiceRecordError(msg
)
3840 self
._log
.info("Create NetworkServiceRecord nsr id %s from nsd_id %s",
3844 nsm_plugin
= self
._ro
_plugin
_selector
.ro_plugin
3845 sdn_account_name
= self
._cloud
_account
_handler
.get_cloud_account_sdn_name(nsr_msg
.cloud_account
)
3847 nsr
= NetworkServiceRecord(self
._dts
,
3855 restart_mode
=restart_mode
,
3856 vlr_handler
=self
._ro
_plugin
_selector
._records
_publisher
._vlr
_pub
_hdlr
3858 self
._nsrs
[nsr_msg
.id] = nsr
3859 nsm_plugin
.create_nsr(nsr_msg
, nsr_msg
.nsd
, key_pairs
)
3863 def delete_nsr(self
, nsr_id
):
3865 Delete NSR with the passed nsr id
3867 del self
._nsrs
[nsr_id
]
3870 def instantiate_ns(self
, nsr_id
, config_xact
):
3871 """ Instantiate an NS instance """
3872 self
._log
.debug("Instantiating Network service id %s", nsr_id
)
3873 if nsr_id
not in self
._nsrs
:
3874 err
= "NSR id %s not found " % nsr_id
3875 self
._log
.error(err
)
3876 raise NetworkServiceRecordError(err
)
3878 nsr
= self
._nsrs
[nsr_id
]
3879 yield from nsr
.nsm_plugin
.instantiate_ns(nsr
, config_xact
)
3882 def update_vnfr(self
, vnfr
):
3883 """Create/Update an VNFR """
3885 vnfr_state
= self
._vnfrs
[vnfr
.id].state
3886 self
._log
.debug("Updating VNFR with state %s: vnfr %s", vnfr_state
, vnfr
)
3888 yield from self
._vnfrs
[vnfr
.id].update_state(vnfr
)
3889 nsr
= self
.find_nsr_for_vnfr(vnfr
.id)
3890 yield from nsr
.update_state()
3892 def find_nsr_for_vnfr(self
, vnfr_id
):
3893 """ Find the NSR which )has the passed vnfr id"""
3894 for nsr
in list(self
.nsrs
.values()):
3895 for vnfr
in list(nsr
.vnfrs
.values()):
3896 if vnfr
.id == vnfr_id
:
3900 def delete_vnfr(self
, vnfr_id
):
3901 """ Delete VNFR with the passed id"""
3902 del self
._vnfrs
[vnfr_id
]
3904 def get_nsd_ref(self
, nsd_id
):
3905 """ Get network service descriptor for the passed nsd_id
3907 nsd
= self
.get_nsd(nsd_id
)
3912 def get_nsr_config(self
, nsd_id
):
3913 xpath
= "C,/nsr:ns-instance-config"
3914 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3916 for result
in results
:
3917 entry
= yield from result
3918 ns_instance_config
= entry
.result
3920 for nsr
in ns_instance_config
.nsr
:
3921 if nsr
.nsd
.id == nsd_id
:
3927 def nsd_unref_by_nsr_id(self
, nsr_id
):
3928 """ Unref the network service descriptor based on NSR id """
3929 self
._log
.debug("NSR Unref called for Nsr Id:%s", nsr_id
)
3930 if nsr_id
in self
._nsrs
:
3931 nsr
= self
._nsrs
[nsr_id
]
3934 nsd
= self
.get_nsd(nsr
.nsd_id
)
3935 self
._log
.debug("Releasing ref on NSD %s held by NSR %s - Curr %d",
3936 nsd
.id, nsr
.id, nsd
.ref_count
)
3938 except NetworkServiceDescriptorError
:
3939 # We store a copy of NSD in NSR and the NSD in nsd-catalog
3944 self
._log
.error("Cannot find NSR with id %s", nsr_id
)
3945 raise NetworkServiceDescriptorUnrefError("No NSR with id" % nsr_id
)
3948 def nsd_unref(self
, nsd_id
):
3949 """ Unref the network service descriptor associated with the id """
3950 nsd
= self
.get_nsd(nsd_id
)
3953 def get_nsd(self
, nsd_id
):
3954 """ Get network service descriptor for the passed nsd_id"""
3955 if nsd_id
not in self
._nsds
:
3956 self
._log
.error("Cannot find NSD id:%s", nsd_id
)
3957 raise NetworkServiceDescriptorError("Cannot find NSD id:%s", nsd_id
)
3959 return self
._nsds
[nsd_id
]
3961 def create_nsd(self
, nsd_msg
):
3962 """ Create a network service descriptor """
3963 self
._log
.debug("Create network service descriptor - %s", nsd_msg
)
3964 if nsd_msg
.id in self
._nsds
:
3965 self
._log
.error("Cannot create NSD %s -NSD ID already exists", nsd_msg
)
3966 raise NetworkServiceDescriptorError("NSD already exists-%s", nsd_msg
.id)
3968 nsd
= NetworkServiceDescriptor(
3975 self
._nsds
[nsd_msg
.id] = nsd
3979 def update_nsd(self
, nsd
):
3980 """ update the Network service descriptor """
3981 self
._log
.debug("Update network service descriptor - %s", nsd
)
3982 if nsd
.id not in self
._nsds
:
3983 self
._log
.debug("No NSD found - creating NSD id = %s", nsd
.id)
3984 self
.create_nsd(nsd
)
3986 self
._log
.debug("Updating NSD id = %s, nsd = %s", nsd
.id, nsd
)
3987 self
._nsds
[nsd
.id].update(nsd
)
3989 def delete_nsd(self
, nsd_id
):
3990 """ Delete the Network service descriptor with the passed id """
3991 self
._log
.debug("Deleting the network service descriptor - %s", nsd_id
)
3992 if nsd_id
not in self
._nsds
:
3993 self
._log
.debug("Delete NSD failed - cannot find nsd-id %s", nsd_id
)
3994 raise NetworkServiceDescriptorNotFound("Cannot find %s", nsd_id
)
3996 if nsd_id
not in self
._nsds
:
3997 self
._log
.debug("Cannot delete NSD id %s reference exists %s",
3999 self
._nsds
[nsd_id
].ref_count
)
4000 raise NetworkServiceDescriptorRefCountExists(
4001 "Cannot delete :%s, ref_count:%s",
4003 self
._nsds
[nsd_id
].ref_count
)
4005 del self
._nsds
[nsd_id
]
4007 def get_vnfd_config(self
, xact
):
4008 vnfd_dts_reg
= self
._vnfd
_dts
_handler
.regh
4009 for cfg
in vnfd_dts_reg
.get_xact_elements(xact
):
4010 if cfg
.id not in self
._vnfds
:
4011 self
.create_vnfd(cfg
)
4013 def get_vnfd(self
, vnfd_id
, xact
):
4014 """ Get virtual network function descriptor for the passed vnfd_id"""
4015 if vnfd_id
not in self
._vnfds
:
4016 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4017 self
.get_vnfd_config(xact
)
4019 if vnfd_id
not in self
._vnfds
:
4020 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4021 raise VnfDescriptorError("Cannot find VNFD id:%s", vnfd_id
)
4023 return self
._vnfds
[vnfd_id
]
4025 def create_vnfd(self
, vnfd
):
4026 """ Create a virtual network function descriptor """
4027 self
._log
.debug("Create virtual network function descriptor - %s", vnfd
)
4028 if vnfd
.id in self
._vnfds
:
4029 self
._log
.error("Cannot create VNFD %s -VNFD ID already exists", vnfd
)
4030 raise VnfDescriptorError("VNFD already exists-%s", vnfd
.id)
4032 self
._vnfds
[vnfd
.id] = vnfd
4033 return self
._vnfds
[vnfd
.id]
4035 def update_vnfd(self
, vnfd
):
4036 """ Update the virtual network function descriptor """
4037 self
._log
.debug("Update virtual network function descriptor- %s", vnfd
)
4040 if vnfd
.id not in self
._vnfds
:
4041 self
._log
.debug("No VNFD found - creating VNFD id = %s", vnfd
.id)
4042 self
.create_vnfd(vnfd
)
4044 self
._log
.debug("Updating VNFD id = %s, vnfd = %s", vnfd
.id, vnfd
)
4045 self
._vnfds
[vnfd
.id] = vnfd
4048 def delete_vnfd(self
, vnfd_id
):
4049 """ Delete the virtual network function descriptor with the passed id """
4050 self
._log
.debug("Deleting the virtual network function descriptor - %s", vnfd_id
)
4051 if vnfd_id
not in self
._vnfds
:
4052 self
._log
.debug("Delete VNFD failed - cannot find vnfd-id %s", vnfd_id
)
4053 raise VnfDescriptorError("Cannot find %s", vnfd_id
)
4055 del self
._vnfds
[vnfd_id
]
4057 def nsd_in_use(self
, nsd_id
):
4058 """ Is the NSD with the passed id in use """
4059 self
._log
.debug("Is this NSD in use - msg:%s", nsd_id
)
4060 if nsd_id
in self
._nsds
:
4061 return self
._nsds
[nsd_id
].in_use()
4065 def publish_nsr(self
, xact
, path
, msg
):
4066 """ Publish a NSR """
4067 self
._log
.debug("Publish NSR with path %s, msg %s",
4069 yield from self
.nsr_handler
.update(xact
, path
, msg
)
4072 def unpublish_nsr(self
, xact
, path
):
4073 """ Un Publish an NSR """
4074 self
._log
.debug("Publishing delete NSR with path %s", path
)
4075 yield from self
.nsr_handler
.delete(path
, xact
)
4077 def vnfr_is_ready(self
, vnfr_id
):
4078 """ VNFR with the id is ready """
4079 self
._log
.debug("VNFR id %s ready", vnfr_id
)
4080 if vnfr_id
not in self
._vnfds
:
4081 err
= "Did not find VNFR ID with id %s" % vnfr_id
4082 self
._log
.critical("err")
4083 raise VirtualNetworkFunctionRecordError(err
)
4084 self
._vnfrs
[vnfr_id
].is_ready()
4087 def get_nsd_refcount(self
, nsd_id
):
4088 """ Get the nsd_list from this NSM"""
4090 def nsd_refcount_xpath(nsd_id
):
4091 """ xpath for ref count entry """
4092 return (NsdRefCountDtsHandler
.XPATH
+
4093 "[rw-nsr:nsd-id-ref = '{}']").format(nsd_id
)
4096 if nsd_id
is None or nsd_id
== "":
4097 for nsd
in self
._nsds
.values():
4098 nsd_msg
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4099 nsd_msg
.nsd_id_ref
= nsd
.id
4100 nsd_msg
.instance_ref_count
= nsd
.ref_count
4101 nsd_list
.append((nsd_refcount_xpath(nsd
.id), nsd_msg
))
4102 elif nsd_id
in self
._nsds
:
4103 nsd_msg
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4104 nsd_msg
.nsd_id_ref
= self
._nsds
[nsd_id
].id
4105 nsd_msg
.instance_ref_count
= self
._nsds
[nsd_id
].ref_count
4106 nsd_list
.append((nsd_refcount_xpath(nsd_id
), nsd_msg
))
4111 def terminate_ns(self
, nsr_id
, xact
):
4113 Terminate network service for the given NSR Id
4116 # Terminate the instances/networks assocaited with this nw service
4117 self
._log
.debug("Terminating the network service %s", nsr_id
)
4118 yield from self
._nsrs
[nsr_id
].terminate()
4121 yield from self
.nsd_unref_by_nsr_id(nsr_id
)
4123 # Unpublish the NSR record
4124 self
._log
.debug("Unpublishing the network service %s", nsr_id
)
4125 yield from self
._nsrs
[nsr_id
].unpublish(xact
)
4127 # Finaly delete the NS instance from this NS Manager
4128 self
._log
.debug("Deletng the network service %s", nsr_id
)
4129 self
.delete_nsr(nsr_id
)
4132 class NsmRecordsPublisherProxy(object):
4133 """ This class provides a publisher interface that allows plugin objects
4134 to publish NSR/VNFR/VLR"""
4136 def __init__(self
, dts
, log
, loop
, nsr_pub_hdlr
, vnfr_pub_hdlr
, vlr_pub_hdlr
):
4140 self
._nsr
_pub
_hdlr
= nsr_pub_hdlr
4141 self
._vlr
_pub
_hdlr
= vlr_pub_hdlr
4142 self
._vnfr
_pub
_hdlr
= vnfr_pub_hdlr
4145 def publish_nsr(self
, xact
, nsr
):
4146 """ Publish an NSR """
4147 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4148 return (yield from self
._nsr
_pub
_hdlr
.update(xact
, path
, nsr
))
4151 def unpublish_nsr(self
, xact
, nsr
):
4152 """ Unpublish an NSR """
4153 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4154 return (yield from self
._nsr
_pub
_hdlr
.delete(xact
, path
))
4157 def publish_vnfr(self
, xact
, vnfr
):
4158 """ Publish an VNFR """
4159 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4160 return (yield from self
._vnfr
_pub
_hdlr
.update(xact
, path
, vnfr
))
4163 def unpublish_vnfr(self
, xact
, vnfr
):
4164 """ Unpublish a VNFR """
4165 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4166 return (yield from self
._vnfr
_pub
_hdlr
.delete(xact
, path
))
4169 def publish_vlr(self
, xact
, vlr
):
4170 """ Publish a VLR """
4171 path
= VirtualLinkRecord
.vlr_xpath(vlr
)
4172 return (yield from self
._vlr
_pub
_hdlr
.update(xact
, path
, vlr
))
4175 def unpublish_vlr(self
, xact
, vlr
):
4176 """ Unpublish a VLR """
4177 path
= VirtualLinkRecord
.vlr_xpath(vlr
)
4178 return (yield from self
._vlr
_pub
_hdlr
.delete(xact
, path
))
4181 class ScalingRpcHandler(mano_dts
.DtsHandler
):
4182 """ The Network service Monitor DTS handler """
4183 SCALE_IN_INPUT_XPATH
= "I,/nsr:exec-scale-in"
4184 SCALE_IN_OUTPUT_XPATH
= "O,/nsr:exec-scale-in"
4186 SCALE_OUT_INPUT_XPATH
= "I,/nsr:exec-scale-out"
4187 SCALE_OUT_OUTPUT_XPATH
= "O,/nsr:exec-scale-out"
4189 ACTION
= Enum('ACTION', 'SCALE_IN SCALE_OUT')
4191 def __init__(self
, log
, dts
, loop
, callback
=None):
4192 super().__init
__(log
, dts
, loop
)
4193 self
.callback
= callback
4194 self
.last_instance_id
= defaultdict(int)
4200 def on_scale_in_prepare(xact_info
, action
, ks_path
, msg
):
4201 assert action
== rwdts
.QueryAction
.RPC
4205 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_IN
)
4207 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleIn
.from_dict({
4208 "instance_id": msg
.instance_id
})
4210 xact_info
.respond_xpath(
4211 rwdts
.XactRspCode
.ACK
,
4212 self
.__class
__.SCALE_IN_OUTPUT_XPATH
,
4215 except Exception as e
:
4216 self
.log
.exception(e
)
4217 xact_info
.respond_xpath(
4218 rwdts
.XactRspCode
.NACK
,
4219 self
.__class
__.SCALE_IN_OUTPUT_XPATH
)
4222 def on_scale_out_prepare(xact_info
, action
, ks_path
, msg
):
4223 assert action
== rwdts
.QueryAction
.RPC
4226 scaling_group
= msg
.scaling_group_name_ref
4227 if not msg
.instance_id
:
4228 last_instance_id
= self
.last_instance_id
[scale_group
]
4229 msg
.instance_id
= last_instance_id
+ 1
4230 self
.last_instance_id
[scale_group
] += 1
4233 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_OUT
)
4235 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleOut
.from_dict({
4236 "instance_id": msg
.instance_id
})
4238 xact_info
.respond_xpath(
4239 rwdts
.XactRspCode
.ACK
,
4240 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
,
4243 except Exception as e
:
4244 self
.log
.exception(e
)
4245 xact_info
.respond_xpath(
4246 rwdts
.XactRspCode
.NACK
,
4247 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
)
4249 scale_in_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4250 on_prepare
=on_scale_in_prepare
)
4251 scale_out_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4252 on_prepare
=on_scale_out_prepare
)
4254 with self
.dts
.group_create() as group
:
4256 xpath
=self
.__class
__.SCALE_IN_INPUT_XPATH
,
4257 handler
=scale_in_hdl
,
4258 flags
=rwdts
.Flag
.PUBLISHER
)
4260 xpath
=self
.__class
__.SCALE_OUT_INPUT_XPATH
,
4261 handler
=scale_out_hdl
,
4262 flags
=rwdts
.Flag
.PUBLISHER
)
4265 class NsmTasklet(rift
.tasklets
.Tasklet
):
4267 The network service manager tasklet
4269 def __init__(self
, *args
, **kwargs
):
4270 super(NsmTasklet
, self
).__init
__(*args
, **kwargs
)
4271 self
.rwlog
.set_category("rw-mano-log")
4272 self
.rwlog
.set_subcategory("nsm")
4277 self
._ro
_plugin
_selector
= None
4278 self
._vnffgmgr
= None
4280 self
._nsr
_handler
= None
4281 self
._vnfr
_pub
_handler
= None
4282 self
._vlr
_pub
_handler
= None
4283 self
._vnfd
_pub
_handler
= None
4284 self
._scale
_cfg
_handler
= None
4286 self
._records
_publisher
_proxy
= None
4289 """ The task start callback """
4290 super(NsmTasklet
, self
).start()
4291 self
.log
.info("Starting NsmTasklet")
4293 self
.log
.debug("Registering with dts")
4294 self
._dts
= rift
.tasklets
.DTS(self
.tasklet_info
,
4295 RwNsmYang
.get_schema(),
4297 self
.on_dts_state_change
)
4299 self
.log
.debug("Created DTS Api GI Object: %s", self
._dts
)
4305 print("Caught Exception in NSM stop:", sys
.exc_info()[0])
4308 def on_instance_started(self
):
4309 """ Task instance started callback """
4310 self
.log
.debug("Got instance started callback")
4314 """ Task init callback """
4315 self
.log
.debug("Got instance started callback")
4317 self
.log
.debug("creating config account handler")
4319 self
._nsr
_pub
_handler
= publisher
.NsrOpDataDtsHandler(self
._dts
, self
.log
, self
.loop
)
4320 yield from self
._nsr
_pub
_handler
.register()
4322 self
._vnfr
_pub
_handler
= publisher
.VnfrPublisherDtsHandler(self
._dts
, self
.log
, self
.loop
)
4323 yield from self
._vnfr
_pub
_handler
.register()
4325 self
._vlr
_pub
_handler
= publisher
.VlrPublisherDtsHandler(self
._dts
, self
.log
, self
.loop
)
4326 yield from self
._vlr
_pub
_handler
.register()
4328 manifest
= self
.tasklet_info
.get_pb_manifest()
4329 use_ssl
= manifest
.bootstrap_phase
.rwsecurity
.use_ssl
4330 ssl_cert
= manifest
.bootstrap_phase
.rwsecurity
.cert
4331 ssl_key
= manifest
.bootstrap_phase
.rwsecurity
.key
4333 self
._vnfd
_pub
_handler
= publisher
.VnfdPublisher(use_ssl
, ssl_cert
, ssl_key
, self
.loop
)
4335 self
._records
_publisher
_proxy
= NsmRecordsPublisherProxy(
4339 self
._nsr
_pub
_handler
,
4340 self
._vnfr
_pub
_handler
,
4341 self
._vlr
_pub
_handler
,
4344 # Register the NSM to receive the nsm plugin
4345 # when cloud account is configured
4346 self
._ro
_plugin
_selector
= cloud
.ROAccountPluginSelector(
4350 self
._records
_publisher
_proxy
,
4352 yield from self
._ro
_plugin
_selector
.register()
4354 self
._cloud
_account
_handler
= cloud
.CloudAccountConfigSubscriber(
4359 yield from self
._cloud
_account
_handler
.register()
4361 self
._vnffgmgr
= rwvnffgmgr
.VnffgMgr(self
._dts
, self
.log
, self
.log_hdl
, self
.loop
)
4362 yield from self
._vnffgmgr
.register()
4364 self
._nsm
= NsManager(
4368 self
._nsr
_pub
_handler
,
4369 self
._vnfr
_pub
_handler
,
4370 self
._vlr
_pub
_handler
,
4371 self
._ro
_plugin
_selector
,
4373 self
._vnfd
_pub
_handler
,
4374 self
._cloud
_account
_handler
4377 yield from self
._nsm
.register()
4381 """ Task run callback """
4385 def on_dts_state_change(self
, state
):
4386 """Take action according to current dts state to transition
4387 application into the corresponding application state
4390 state - current dts state
4393 rwdts
.State
.INIT
: rwdts
.State
.REGN_COMPLETE
,
4394 rwdts
.State
.CONFIG
: rwdts
.State
.RUN
,
4398 rwdts
.State
.INIT
: self
.init
,
4399 rwdts
.State
.RUN
: self
.run
,
4402 # Transition application to next state
4403 handler
= handlers
.get(state
, None)
4404 if handler
is not None:
4405 yield from handler()
4407 # Transition dts to next state
4408 next_state
= switch
.get(state
, None)
4409 if next_state
is not None:
4410 self
.log
.debug("Changing state to %s", next_state
)
4411 self
._dts
.handle
.set_state(next_state
)