2 # Copyright 2016 RIFT.IO Inc
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
17 # vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
20 import ncclient
.asyncio_manager
32 from collections
import deque
33 from collections
import defaultdict
37 gi
.require_version('RwYang', '1.0')
38 gi
.require_version('RwNsdYang', '1.0')
39 gi
.require_version('RwDts', '1.0')
40 gi
.require_version('RwNsmYang', '1.0')
41 gi
.require_version('RwNsrYang', '1.0')
42 gi
.require_version('RwTypes', '1.0')
43 gi
.require_version('RwVlrYang', '1.0')
44 gi
.require_version('RwVnfrYang', '1.0')
45 from gi
.repository
import (
61 import rift
.mano
.ncclient
62 import rift
.mano
.config_data
.config
63 import rift
.mano
.dts
as mano_dts
65 from . import rwnsm_conman
as conman
67 from . import publisher
69 from . import config_value_pool
70 from . import rwvnffgmgr
71 from . import scale_group
74 class NetworkServiceRecordState(Enum
):
75 """ Network Service Record State """
79 VNFFG_INIT_PHASE
= 104
85 VL_TERMINATE_PHASE
= 111
86 VNF_TERMINATE_PHASE
= 112
87 VNFFG_TERMINATE_PHASE
= 113
94 class NetworkServiceRecordError(Exception):
95 """ Network Service Record Error """
99 class NetworkServiceDescriptorError(Exception):
100 """ Network Service Descriptor Error """
104 class VirtualNetworkFunctionRecordError(Exception):
105 """ Virtual Network Function Record Error """
109 class NetworkServiceDescriptorNotFound(Exception):
110 """ Cannot find Network Service Descriptor"""
114 class NetworkServiceDescriptorRefCountExists(Exception):
115 """ Network Service Descriptor reference count exists """
119 class NetworkServiceDescriptorUnrefError(Exception):
120 """ Failed to unref a network service descriptor """
124 class NsrInstantiationFailed(Exception):
125 """ Failed to instantiate network service """
129 class VnfInstantiationFailed(Exception):
130 """ Failed to instantiate virtual network function"""
134 class VnffgInstantiationFailed(Exception):
135 """ Failed to instantiate virtual network function"""
139 class VnfDescriptorError(Exception):
140 """Failed to instantiate virtual network function"""
144 class ScalingOperationError(Exception):
148 class ScaleGroupMissingError(Exception):
152 class PlacementGroupError(Exception):
156 class NsrNsdUpdateError(Exception):
160 class NsrVlUpdateError(NsrNsdUpdateError
):
164 class VlRecordState(Enum
):
165 """ VL Record State """
167 INSTANTIATION_PENDING
= 102
169 TERMINATE_PENDING
= 104
174 class VnffgRecordState(Enum
):
175 """ VNFFG Record State """
177 INSTANTIATION_PENDING
= 102
179 TERMINATE_PENDING
= 104
184 class VnffgRecord(object):
185 """ Vnffg Records class"""
188 def __init__(self
, dts
, log
, loop
, vnffgmgr
, nsr
, nsr_name
, vnffgd_msg
, sdn_account_name
):
193 self
._vnffgmgr
= vnffgmgr
195 self
._nsr
_name
= nsr_name
196 self
._vnffgd
_msg
= vnffgd_msg
197 if sdn_account_name
is None:
198 self
._sdn
_account
_name
= ''
200 self
._sdn
_account
_name
= sdn_account_name
202 self
._vnffgr
_id
= str(uuid
.uuid4())
203 self
._vnffgr
_rsp
_id
= list()
204 self
._vnffgr
_state
= VnffgRecordState
.INIT
209 return self
._vnffgr
_id
213 """ state of this VNF """
214 return self
._vnffgr
_state
216 def fetch_vnffgr(self
):
218 Get VNFFGR message to be published
221 if self
._vnffgr
_state
== VnffgRecordState
.INIT
:
222 vnffgr_dict
= {"id": self
._vnffgr
_id
,
223 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
224 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
225 "sdn_account": self
._sdn
_account
_name
,
226 "operational_status": 'init',
228 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
229 elif self
._vnffgr
_state
== VnffgRecordState
.TERMINATED
:
230 vnffgr_dict
= {"id": self
._vnffgr
_id
,
231 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
232 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
233 "sdn_account": self
._sdn
_account
_name
,
234 "operational_status": 'terminated',
236 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
239 vnffgr
= self
._vnffgmgr
.fetch_vnffgr(self
._vnffgr
_id
)
241 self
._log
.exception("Fetching VNFFGR for VNFFG with id %s failed", self
._vnffgr
_id
)
242 self
._vnffgr
_state
= VnffgRecordState
.FAILED
243 vnffgr_dict
= {"id": self
._vnffgr
_id
,
244 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
245 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
246 "sdn_account": self
._sdn
_account
_name
,
247 "operational_status": 'failed',
249 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
254 def vnffgr_create_msg(self
):
255 """ Virtual Link Record message for Creating VLR in VNS """
256 vnffgr_dict
= {"id": self
._vnffgr
_id
,
257 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
258 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
259 "sdn_account": self
._sdn
_account
_name
,
261 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
262 for rsp
in self
._vnffgd
_msg
.rsp
:
263 vnffgr_rsp
= vnffgr
.rsp
.add()
264 vnffgr_rsp
.id = str(uuid
.uuid4())
265 vnffgr_rsp
.name
= self
._nsr
.name
+ '.' + rsp
.name
266 self
._vnffgr
_rsp
_id
.append(vnffgr_rsp
.id)
267 vnffgr_rsp
.vnffgd_rsp_id_ref
= rsp
.id
268 vnffgr_rsp
.vnffgd_rsp_name_ref
= rsp
.name
269 for rsp_cp_ref
in rsp
.vnfd_connection_point_ref
:
270 vnfd
= [vnfr
.vnfd
for vnfr
in self
._nsr
.vnfrs
.values() if vnfr
.vnfd
.id == rsp_cp_ref
.vnfd_id_ref
]
271 self
._log
.debug("VNFD message during VNFFG instantiation is %s",vnfd
)
272 if len(vnfd
) > 0 and vnfd
[0].has_field('service_function_type'):
273 self
._log
.debug("Service Function Type for VNFD ID %s is %s",rsp_cp_ref
.vnfd_id_ref
, vnfd
[0].service_function_type
)
275 self
._log
.error("Service Function Type not available for VNFD ID %s; Skipping in chain",rsp_cp_ref
.vnfd_id_ref
)
278 vnfr_cp_ref
= vnffgr_rsp
.vnfr_connection_point_ref
.add()
279 vnfr_cp_ref
.member_vnf_index_ref
= rsp_cp_ref
.member_vnf_index_ref
280 vnfr_cp_ref
.hop_number
= rsp_cp_ref
.order
281 vnfr_cp_ref
.vnfd_id_ref
=rsp_cp_ref
.vnfd_id_ref
282 vnfr_cp_ref
.service_function_type
= vnfd
[0].service_function_type
283 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
284 if (nsr_vnfr
.vnfd
.id == vnfr_cp_ref
.vnfd_id_ref
and
285 nsr_vnfr
.member_vnf_index
== vnfr_cp_ref
.member_vnf_index_ref
):
286 vnfr_cp_ref
.vnfr_id_ref
= nsr_vnfr
.id
287 vnfr_cp_ref
.vnfr_name_ref
= nsr_vnfr
.name
288 vnfr_cp_ref
.vnfr_connection_point_ref
= rsp_cp_ref
.vnfd_connection_point_ref
290 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
291 self
._log
.debug(" Received VNFR is %s", vnfr
)
292 while vnfr
.operational_status
!= 'running':
293 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
294 if vnfr
.operational_status
== 'failed':
295 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
296 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self
.id, vnfr
.id))
297 yield from asyncio
.sleep(2, loop
=self
._loop
)
298 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
299 self
._log
.debug("Received VNFR is %s", vnfr
)
301 vnfr_cp_ref
.connection_point_params
.mgmt_address
= vnfr
.mgmt_interface
.ip_address
302 for cp
in vnfr
.connection_point
:
303 if cp
.name
== vnfr_cp_ref
.vnfr_connection_point_ref
:
304 vnfr_cp_ref
.connection_point_params
.port_id
= cp
.connection_point_id
305 vnfr_cp_ref
.connection_point_params
.name
= self
._nsr
.name
+ '.' + cp
.name
306 for vdu
in vnfr
.vdur
:
307 for ext_intf
in vdu
.external_interface
:
308 if ext_intf
.name
== vnfr_cp_ref
.vnfr_connection_point_ref
:
309 vnfr_cp_ref
.connection_point_params
.vm_id
= vdu
.vim_id
310 self
._log
.debug("VIM ID for CP %s in VNFR %s is %s",cp
.name
,nsr_vnfr
.id,
311 vnfr_cp_ref
.connection_point_params
.vm_id
)
314 vnfr_cp_ref
.connection_point_params
.address
= cp
.ip_address
315 vnfr_cp_ref
.connection_point_params
.port
= VnffgRecord
.SFF_DP_PORT
317 for vnffgd_classifier
in self
._vnffgd
_msg
.classifier
:
318 _rsp
= [rsp
for rsp
in vnffgr
.rsp
if rsp
.vnffgd_rsp_id_ref
== vnffgd_classifier
.rsp_id_ref
]
320 rsp_id_ref
= _rsp
[0].id
321 rsp_name
= _rsp
[0].name
323 self
._log
.error("RSP with ID %s not found during classifier creation for classifier id %s",vnffgd_classifier
.rsp_id_ref
,vnffgd_classifier
.id)
325 vnffgr_classifier
= vnffgr
.classifier
.add()
326 vnffgr_classifier
.id = vnffgd_classifier
.id
327 vnffgr_classifier
.name
= self
._nsr
.name
+ '.' + vnffgd_classifier
.name
328 _rsp
[0].classifier_name
= vnffgr_classifier
.name
329 vnffgr_classifier
.rsp_id_ref
= rsp_id_ref
330 vnffgr_classifier
.rsp_name
= rsp_name
331 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
332 if (nsr_vnfr
.vnfd
.id == vnffgd_classifier
.vnfd_id_ref
and
333 nsr_vnfr
.member_vnf_index
== vnffgd_classifier
.member_vnf_index_ref
):
334 vnffgr_classifier
.vnfr_id_ref
= nsr_vnfr
.id
335 vnffgr_classifier
.vnfr_name_ref
= nsr_vnfr
.name
336 vnffgr_classifier
.vnfr_connection_point_ref
= vnffgd_classifier
.vnfd_connection_point_ref
338 if nsr_vnfr
.vnfd
.service_function_chain
== 'CLASSIFIER':
339 vnffgr_classifier
.sff_name
= nsr_vnfr
.name
341 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
342 self
._log
.debug(" Received VNFR is %s", vnfr
)
343 while vnfr
.operational_status
!= 'running':
344 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
345 if vnfr
.operational_status
== 'failed':
346 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
347 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self
.id, vnfr
.id))
348 yield from asyncio
.sleep(2, loop
=self
._loop
)
349 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
350 self
._log
.debug("Received VNFR is %s", vnfr
)
352 for cp
in vnfr
.connection_point
:
353 if cp
.name
== vnffgr_classifier
.vnfr_connection_point_ref
:
354 vnffgr_classifier
.port_id
= cp
.connection_point_id
355 vnffgr_classifier
.ip_address
= cp
.ip_address
356 for vdu
in vnfr
.vdur
:
357 for ext_intf
in vdu
.external_interface
:
358 if ext_intf
.name
== vnffgr_classifier
.vnfr_connection_point_ref
:
359 vnffgr_classifier
.vm_id
= vdu
.vim_id
360 self
._log
.debug("VIM ID for CP %s in VNFR %s is %s",cp
.name
,nsr_vnfr
.id,
361 vnfr_cp_ref
.connection_point_params
.vm_id
)
364 self
._log
.info("VNFFGR msg to be sent is %s", vnffgr
)
368 def vnffgr_nsr_sff_list(self
):
369 """ SFF List for VNFR """
371 sf_list
= [nsr_vnfr
.name
for nsr_vnfr
in self
._nsr
.vnfrs
.values() if nsr_vnfr
.vnfd
.service_function_chain
== 'SF']
373 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
374 if (nsr_vnfr
.vnfd
.service_function_chain
== 'CLASSIFIER' or nsr_vnfr
.vnfd
.service_function_chain
== 'SFF'):
375 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
376 self
._log
.debug(" Received VNFR is %s", vnfr
)
377 while vnfr
.operational_status
!= 'running':
378 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
379 if vnfr
.operational_status
== 'failed':
380 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
381 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self
.id, vnfr
.id))
382 yield from asyncio
.sleep(2, loop
=self
._loop
)
383 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
384 self
._log
.debug("Received VNFR is %s", vnfr
)
386 sff
= RwsdnYang
.VNFFGSff()
387 sff_list
[nsr_vnfr
.vnfd
.id] = sff
388 sff
.name
= nsr_vnfr
.name
389 sff
.function_type
= nsr_vnfr
.vnfd
.service_function_chain
391 sff
.mgmt_address
= vnfr
.mgmt_interface
.ip_address
392 sff
.mgmt_port
= VnffgRecord
.SFF_MGMT_PORT
393 for cp
in vnfr
.connection_point
:
394 sff_dp
= sff
.dp_endpoints
.add()
395 sff_dp
.name
= self
._nsr
.name
+ '.' + cp
.name
396 sff_dp
.address
= cp
.ip_address
397 sff_dp
.port
= VnffgRecord
.SFF_DP_PORT
398 if nsr_vnfr
.vnfd
.service_function_chain
== 'SFF':
399 for sf_name
in sf_list
:
400 _sf
= sff
.vnfr_list
.add()
401 _sf
.vnfr_name
= sf_name
406 def instantiate(self
):
407 """ Instantiate this VNFFG """
409 self
._log
.info("Instaniating VNFFGR with vnffgd %s",
413 vnffgr_request
= yield from self
.vnffgr_create_msg()
414 vnffg_sff_list
= yield from self
.vnffgr_nsr_sff_list()
417 vnffgr
= self
._vnffgmgr
.create_vnffgr(vnffgr_request
,self
._vnffgd
_msg
.classifier
,vnffg_sff_list
)
418 except Exception as e
:
419 self
._log
.exception("VNFFG instantiation failed: %s", str(e
))
420 self
._vnffgr
_state
= VnffgRecordState
.FAILED
421 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFFGR %s failure" % (self
.id, vnffgr_request
.id))
423 self
._vnffgr
_state
= VnffgRecordState
.INSTANTIATION_PENDING
425 self
._log
.info("Instantiated VNFFGR :%s", vnffgr
)
426 self
._vnffgr
_state
= VnffgRecordState
.ACTIVE
428 self
._log
.info("Invoking update_state to update NSR state for NSR ID: %s", self
._nsr
.id)
429 yield from self
._nsr
.update_state()
431 def vnffgr_in_vnffgrm(self
):
432 """ Is there a VNFR record in VNFM """
433 if (self
._vnffgr
_state
== VnffgRecordState
.ACTIVE
or
434 self
._vnffgr
_state
== VnffgRecordState
.INSTANTIATION_PENDING
or
435 self
._vnffgr
_state
== VnffgRecordState
.FAILED
):
442 """ Terminate this VNFFGR """
443 if not self
.vnffgr_in_vnffgrm():
444 self
._log
.error("Ignoring terminate request for id %s in state %s",
445 self
.id, self
._vnffgr
_state
)
448 self
._log
.info("Terminating VNFFGR id:%s", self
.id)
449 self
._vnffgr
_state
= VnffgRecordState
.TERMINATE_PENDING
451 self
._vnffgmgr
.terminate_vnffgr(self
._vnffgr
_id
)
453 self
._vnffgr
_state
= VnffgRecordState
.TERMINATED
454 self
._log
.debug("Terminated VNFFGR id:%s", self
.id)
457 class VirtualLinkRecord(object):
458 """ Virtual Link Records class"""
461 def create_record(dts
, log
, loop
, nsr_name
, vld_msg
, cloud_account_name
, om_datacenter
, ip_profile
, nsr_id
, restart_mode
=False):
462 """Creates a new VLR object based on the given data.
464 If restart mode is enabled, then we look for existing records in the
465 DTS and create a VLR records using the exiting data(ID)
470 vlr_obj
= VirtualLinkRecord(
483 res_iter
= yield from dts
.query_read(
484 "D,/vlr:vlr-catalog/vlr:vlr",
485 rwdts
.XactFlag
.MERGE
)
488 response
= yield from fut
489 vlr
= response
.result
491 # Check if the record is already present, if so use the ID of
492 # the existing record. Since the name of the record is uniquely
493 # formed we can use it as a search key!
494 if vlr
.name
== vlr_obj
.name
:
495 vlr_obj
.reset_id(vlr
.id)
500 def __init__(self
, dts
, log
, loop
, nsr_name
, vld_msg
, cloud_account_name
, om_datacenter
, ip_profile
, nsr_id
):
504 self
._nsr
_name
= nsr_name
505 self
._vld
_msg
= vld_msg
506 self
._cloud
_account
_name
= cloud_account_name
507 self
._om
_datacenter
_name
= om_datacenter
508 self
._assigned
_subnet
= None
509 self
._nsr
_id
= nsr_id
510 self
._ip
_profile
= ip_profile
511 self
._vlr
_id
= str(uuid
.uuid4())
512 self
._state
= VlRecordState
.INIT
513 self
._prev
_state
= None
517 """ path for this object """
518 return "D,/vlr:vlr-catalog/vlr:vlr[vlr:id = '{}']".format(self
._vlr
_id
)
527 """ Get NSR name for this VL """
532 """ Virtual Link Desciptor """
536 def assigned_subnet(self
):
537 """ Subnet assigned to this VL"""
538 return self
._assigned
_subnet
543 Get the name for this VLR.
544 VLR name is "nsr name:VLD name"
546 if self
.vld_msg
.vim_network_name
:
547 return self
.vld_msg
.vim_network_name
548 elif self
.vld_msg
.name
== "multisite":
549 # This is a temporary hack to identify manually provisioned inter-site network
550 return self
.vld_msg
.name
552 return self
._nsr
_name
+ "." + self
.vld_msg
.name
555 def cloud_account_name(self
):
556 """ Cloud account that this VLR should be created in """
557 return self
._cloud
_account
_name
560 def om_datacenter_name(self
):
561 """ Datacenter that this VLR should be created in """
562 return self
._om
_datacenter
_name
566 """ Get the VLR path from VLR """
567 return (VirtualLinkRecord
.XPATH
+ "[vlr:id = '{}']").format(vlr
.id)
575 def state(self
, value
):
576 """ VLR set state """
580 def prev_state(self
):
581 """ VLR previous state """
582 return self
._prev
_state
585 def prev_state(self
, value
):
586 """ VLR set previous state """
587 self
._prev
_state
= value
591 """ Virtual Link Record message for Creating VLR in VNS """
592 vld_fields
= ["short_name",
600 vld_copy_dict
= {k
: v
for k
, v
in self
.vld_msg
.as_dict().items()
603 vlr_dict
= {"id": self
._vlr
_id
,
604 "nsr_id_ref": self
._nsr
_id
,
605 "vld_ref": self
.vld_msg
.id,
607 "cloud_account": self
.cloud_account_name
,
608 "om_datacenter": self
.om_datacenter_name
,
611 if self
._ip
_profile
and self
._ip
_profile
.has_field('ip_profile_params'):
612 vlr_dict
['ip_profile_params' ] = self
._ip
_profile
.ip_profile_params
.as_dict()
614 vlr_dict
.update(vld_copy_dict
)
615 vlr
= RwVlrYang
.YangData_Vlr_VlrCatalog_Vlr
.from_dict(vlr_dict
)
618 def reset_id(self
, vlr_id
):
619 self
._vlr
_id
= vlr_id
621 def create_nsr_vlr_msg(self
, vnfrs
):
622 """ The VLR message"""
623 nsr_vlr
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vlr()
624 nsr_vlr
.vlr_ref
= self
._vlr
_id
625 nsr_vlr
.assigned_subnet
= self
.assigned_subnet
626 nsr_vlr
.cloud_account
= self
.cloud_account_name
627 nsr_vlr
.om_datacenter
= self
.om_datacenter_name
629 for conn
in self
.vld_msg
.vnfd_connection_point_ref
:
631 if (vnfr
.vnfd
.id == conn
.vnfd_id_ref
and
632 vnfr
.member_vnf_index
== conn
.member_vnf_index_ref
and
633 self
.cloud_account_name
== vnfr
.cloud_account_name
and
634 self
.om_datacenter_name
== vnfr
.om_datacenter_name
):
635 cp_entry
= nsr_vlr
.vnfr_connection_point_ref
.add()
636 cp_entry
.vnfr_id
= vnfr
.id
637 cp_entry
.connection_point
= conn
.vnfd_connection_point_ref
642 def instantiate(self
):
643 """ Instantiate this VL """
645 self
._log
.debug("Instaniating VLR key %s, vld %s",
646 self
.xpath
, self
._vld
_msg
)
648 self
._state
= VlRecordState
.INSTANTIATION_PENDING
649 self
._log
.debug("Executing VL create path:%s msg:%s",
650 self
.xpath
, self
.vlr_msg
)
652 with self
._dts
.transaction(flags
=0) as xact
:
653 block
= xact
.block_create()
654 block
.add_query_create(self
.xpath
, self
.vlr_msg
)
655 self
._log
.debug("Executing VL create path:%s msg:%s",
656 self
.xpath
, self
.vlr_msg
)
657 res_iter
= yield from block
.execute(now
=True)
663 self
._state
= VlRecordState
.FAILED
664 raise NsrInstantiationFailed("Failed NS %s instantiation due to empty response" % self
.id)
666 if vlr
.operational_status
== 'failed':
667 self
._log
.debug("NS Id:%s VL creation failed for vlr id %s", self
.id, vlr
.id)
668 self
._state
= VlRecordState
.FAILED
669 raise NsrInstantiationFailed("Failed VL %s instantiation (%s)" % (vlr
.id, vlr
.operational_status_details
))
671 self
._log
.info("Instantiated VL with xpath %s and vlr:%s",
673 self
._state
= VlRecordState
.ACTIVE
674 self
._assigned
_subnet
= vlr
.assigned_subnet
676 def vlr_in_vns(self
):
677 """ Is there a VLR record in VNS """
678 if (self
._state
== VlRecordState
.ACTIVE
or
679 self
._state
== VlRecordState
.INSTANTIATION_PENDING
or
680 self
._state
== VlRecordState
.TERMINATE_PENDING
or
681 self
._state
== VlRecordState
.FAILED
):
688 """ Terminate this VL """
689 if not self
.vlr_in_vns():
690 self
._log
.debug("Ignoring terminate request for id %s in state %s",
691 self
.id, self
._state
)
694 self
._log
.debug("Terminating VL id:%s", self
.id)
695 self
._state
= VlRecordState
.TERMINATE_PENDING
697 with self
._dts
.transaction(flags
=0) as xact
:
698 block
= xact
.block_create()
699 block
.add_query_delete(self
.xpath
)
700 yield from block
.execute(flags
=0, now
=True)
702 self
._state
= VlRecordState
.TERMINATED
703 self
._log
.debug("Terminated VL id:%s", self
.id)
706 class VnfRecordState(Enum
):
707 """ Vnf Record State """
709 INSTANTIATION_PENDING
= 102
711 TERMINATE_PENDING
= 104
716 class VirtualNetworkFunctionRecord(object):
717 """ Virtual Network Function Record class"""
718 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
722 def create_record(dts
, log
, loop
, vnfd
, const_vnfd_msg
, nsd_id
, nsr_name
,
723 cloud_account_name
, om_datacenter_name
, nsr_id
, group_name
, group_instance_id
,
724 placement_groups
, restart_mode
=False):
725 """Creates a new VNFR object based on the given data.
727 If restart mode is enabled, then we look for existing records in the
728 DTS and create a VNFR records using the exiting data(ID)
731 VirtualNetworkFunctionRecord
733 vnfr_obj
= VirtualNetworkFunctionRecord(
747 restart_mode
=restart_mode
)
750 res_iter
= yield from dts
.query_read(
751 "D,/vnfr:vnfr-catalog/vnfr:vnfr",
752 rwdts
.XactFlag
.MERGE
)
755 response
= yield from fut
756 vnfr
= response
.result
758 if vnfr
.name
== vnfr_obj
.name
:
759 vnfr_obj
.reset_id(vnfr
.id)
776 group_instance_id
=None,
777 placement_groups
= [],
778 restart_mode
= False):
783 self
._const
_vnfd
_msg
= const_vnfd_msg
784 self
._nsd
_id
= nsd_id
785 self
._nsr
_name
= nsr_name
786 self
._nsr
_id
= nsr_id
787 self
._cloud
_account
_name
= cloud_account_name
788 self
._om
_datacenter
_name
= om_datacenter_name
789 self
._group
_name
= group_name
790 self
._group
_instance
_id
= group_instance_id
791 self
._placement
_groups
= placement_groups
792 self
._config
_status
= NsrYang
.ConfigStates
.INIT
794 self
._prev
_state
= VnfRecordState
.INIT
795 self
._state
= VnfRecordState
.INIT
796 self
._state
_failed
_reason
= None
798 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
801 self
._vnfr
_id
= str(uuid
.uuid4())
803 self
._vnfr
_msg
= self
.create_vnfr_msg()
804 self
._log
.debug("Set VNFR {} config type to {}".
805 format(self
.name
, self
.config_type
))
806 self
.restart_mode
= restart_mode
809 if group_name
is None and group_instance_id
is not None:
810 raise ValueError("Group instance id must not be provided with an empty group name")
820 return "D,/vnfr:vnfr-catalog/vnfr:vnfr[vnfr:id = '{}']".format(self
.id)
825 return self
._vnfr
_msg
828 def const_vnfr_msg(self
):
830 return RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_ConstituentVnfrRef(vnfr_id
=self
.id,cloud_account
=self
.cloud_account_name
,om_datacenter
=self
._om
_datacenter
_name
)
838 def cloud_account_name(self
):
839 """ Cloud account that this VNF should be created in """
840 return self
._cloud
_account
_name
843 def om_datacenter_name(self
):
844 """ Datacenter that this VNF should be created in """
845 return self
._om
_datacenter
_name
850 """ Is this VNF actve """
851 return True if self
._state
== VnfRecordState
.ACTIVE
else False
855 """ state of this VNF """
859 def state_failed_reason(self
):
860 """ Error message in case this VNF is in failed state """
861 return self
._state
_failed
_reason
864 def member_vnf_index(self
):
865 """ Member VNF index """
866 return self
._const
_vnfd
_msg
.member_vnf_index
871 return self
._nsr
_name
875 """ Name of this VNFR """
876 if self
._name
is not None:
879 name_tags
= [self
._nsr
_name
]
881 if self
._group
_name
is not None:
882 name_tags
.append(self
._group
_name
)
884 if self
._group
_instance
_id
is not None:
885 name_tags
.append(str(self
._group
_instance
_id
))
887 name_tags
.extend([self
.vnfd
.name
, str(self
.member_vnf_index
)])
889 self
._name
= "__".join(name_tags
)
894 def vnfr_xpath(vnfr
):
895 """ Get the VNFR path from VNFR """
896 return (VirtualNetworkFunctionRecord
.XPATH
+ "[vnfr:id = '{}']").format(vnfr
.id)
899 def config_type(self
):
900 cfg_types
= ['netconf', 'juju', 'script']
901 for method
in cfg_types
:
902 if self
._vnfd
.vnf_configuration
.has_field(method
):
907 def config_status(self
):
908 """Return the config status as YANG ENUM string"""
909 self
._log
.debug("Map VNFR {} config status {} ({})".
910 format(self
.name
, self
._config
_status
, self
.config_type
))
911 if self
.config_type
== 'none':
912 return 'config_not_needed'
913 elif self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
915 elif self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
920 def set_state(self
, state
):
921 """ set the state of this object """
922 self
._prev
_state
= self
._state
925 def reset_id(self
, vnfr_id
):
926 self
._vnfr
_id
= vnfr_id
927 self
._vnfr
_msg
= self
.create_vnfr_msg()
930 self
.config_store
.merge_vnfd_config(
933 self
.member_vnf_index
,
936 def create_vnfr_msg(self
):
937 """ VNFR message for this VNFR """
945 vnfd_copy_dict
= {k
: v
for k
, v
in self
._vnfd
.as_dict().items() if k
in vnfd_fields
}
948 "nsr_id_ref": self
._nsr
_id
,
949 "vnfd_ref": self
.vnfd
.id,
951 "cloud_account": self
._cloud
_account
_name
,
952 "om_datacenter": self
._om
_datacenter
_name
,
953 "config_status": self
.config_status
955 vnfr_dict
.update(vnfd_copy_dict
)
957 vnfr
= RwVnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr
.from_dict(vnfr_dict
)
958 vnfr
.member_vnf_index_ref
= self
.member_vnf_index
959 vnfr
.vnf_configuration
.from_dict(self
._vnfd
.vnf_configuration
.as_dict())
961 if self
._vnfd
.mgmt_interface
.has_field("port"):
962 vnfr
.mgmt_interface
.port
= self
._vnfd
.mgmt_interface
.port
964 for group_info
in self
._placement
_groups
:
965 group
= vnfr
.placement_groups_info
.add()
966 group
.from_dict(group_info
.as_dict())
968 # UI expects the monitoring param field to exist
969 vnfr
.monitoring_param
= []
971 self
._log
.debug("Get vnfr_msg for VNFR {} : {}".format(self
.name
, vnfr
))
975 def update_vnfm(self
):
976 self
._log
.debug("Send an update to VNFM for VNFR {} with {}".
977 format(self
.name
, self
.vnfr_msg
))
978 yield from self
._dts
.query_update(
980 rwdts
.XactFlag
.TRACE
,
984 def get_config_status(self
):
985 """Return the config status as YANG ENUM"""
986 return self
._config
_status
989 def set_config_status(self
, status
):
991 def status_to_string(status
):
993 NsrYang
.ConfigStates
.INIT
: 'init',
994 NsrYang
.ConfigStates
.CONFIGURING
: 'configuring',
995 NsrYang
.ConfigStates
.CONFIG_NOT_NEEDED
: 'config_not_needed',
996 NsrYang
.ConfigStates
.CONFIGURED
: 'configured',
997 NsrYang
.ConfigStates
.FAILED
: 'failed',
1000 return status_dc
[status
]
1002 self
._log
.debug("Update VNFR {} from {} ({}) to {}".
1003 format(self
.name
, self
._config
_status
,
1004 self
.config_type
, status
))
1005 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1006 self
._log
.error("Updating already configured VNFR {}".
1010 if self
._config
_status
!= status
:
1012 self
._config
_status
= status
1013 # I don't think this is used. Original implementor can check.
1014 # Caused Exception, so corrected it by status_to_string
1015 # But not sure whats the use of this variable?
1016 self
.vnfr_msg
.config_status
= status_to_string(status
)
1017 except Exception as e
:
1018 self
._log
.error("Exception=%s", str(e
))
1021 self
._log
.debug("Updated VNFR {} status to {}".format(self
.name
, status
))
1023 if self
._config
_status
!= NsrYang
.ConfigStates
.INIT
:
1025 # Publish only after VNFM has the VNFR created
1026 yield from self
.update_vnfm()
1027 except Exception as e
:
1028 self
._log
.error("Exception updating VNFM with new status {} of VNFR {}: {}".
1029 format(status
, self
.name
, e
))
1030 self
._log
.exception(e
)
1032 def is_configured(self
):
1033 if self
.config_type
== 'none':
1036 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1042 def instantiate(self
, nsr
):
1043 """ Instantiate this VNFR"""
1045 self
._log
.debug("Instaniating VNFR key %s, vnfd %s",
1046 self
.xpath
, self
._vnfd
)
1048 self
._log
.debug("Create VNF with xpath %s and vnfr %s",
1049 self
.xpath
, self
.vnfr_msg
)
1051 self
.set_state(VnfRecordState
.INSTANTIATION_PENDING
)
1053 def find_vlr_for_cp(conn
):
1054 """ Find VLR for the given connection point """
1055 for vlr
in nsr
.vlrs
:
1056 for vnfd_cp
in vlr
.vld_msg
.vnfd_connection_point_ref
:
1057 if (vnfd_cp
.vnfd_id_ref
== self
._vnfd
.id and
1058 vnfd_cp
.vnfd_connection_point_ref
== conn
.name
and
1059 vnfd_cp
.member_vnf_index_ref
== self
.member_vnf_index
and
1060 vlr
.cloud_account_name
== self
.cloud_account_name
):
1061 self
._log
.debug("Found VLR for cp_name:%s and vnf-index:%d",
1062 conn
.name
, self
.member_vnf_index
)
1066 # For every connection point in the VNFD fill in the identifier
1067 for conn_p
in self
._vnfd
.connection_point
:
1068 cpr
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_ConnectionPoint()
1069 cpr
.name
= conn_p
.name
1070 cpr
.type_yang
= conn_p
.type_yang
1071 vlr_ref
= find_vlr_for_cp(conn_p
)
1073 msg
= "Failed to find VLR for cp = %s" % conn_p
.name
1074 self
._log
.debug("%s", msg
)
1075 # raise VirtualNetworkFunctionRecordError(msg)
1078 cpr
.vlr_ref
= vlr_ref
.id
1079 self
.vnfr_msg
.connection_point
.append(cpr
)
1080 self
._log
.debug("Connection point [%s] added, vnf id=%s vnfd id=%s",
1081 cpr
, self
.vnfr_msg
.id, self
.vnfr_msg
.vnfd_ref
)
1083 if not self
.restart_mode
:
1084 yield from self
._dts
.query_create(self
.xpath
,
1088 yield from self
._dts
.query_update(self
.xpath
,
1092 self
._log
.info("Created VNF with xpath %s and vnfr %s",
1093 self
.xpath
, self
.vnfr_msg
)
1095 self
._log
.info("Instantiated VNFR with xpath %s and vnfd %s, vnfr %s",
1096 self
.xpath
, self
._vnfd
, self
.vnfr_msg
)
1099 def update_state(self
, vnfr_msg
):
1100 """ Update this VNFR"""
1101 if vnfr_msg
.operational_status
== "running":
1102 if self
.vnfr_msg
.operational_status
!= "running":
1103 yield from self
.is_active()
1104 elif vnfr_msg
.operational_status
== "failed":
1105 yield from self
.instantiation_failed(failed_reason
=vnfr_msg
.operational_status_details
)
1108 def is_active(self
):
1109 """ This VNFR is active """
1110 self
._log
.debug("VNFR %s is active", self
._vnfr
_id
)
1111 self
.set_state(VnfRecordState
.ACTIVE
)
1114 def instantiation_failed(self
, failed_reason
=None):
1115 """ This VNFR instantiation failed"""
1116 self
._log
.error("VNFR %s instantiation failed", self
._vnfr
_id
)
1117 self
.set_state(VnfRecordState
.FAILED
)
1118 self
._state
_failed
_reason
= failed_reason
1120 def vnfr_in_vnfm(self
):
1121 """ Is there a VNFR record in VNFM """
1122 if (self
._state
== VnfRecordState
.ACTIVE
or
1123 self
._state
== VnfRecordState
.INSTANTIATION_PENDING
or
1124 self
._state
== VnfRecordState
.FAILED
):
1130 def terminate(self
):
1131 """ Terminate this VNF """
1132 if not self
.vnfr_in_vnfm():
1133 self
._log
.debug("Ignoring terminate request for id %s in state %s",
1134 self
.id, self
._state
)
1137 self
._log
.debug("Terminating VNF id:%s", self
.id)
1138 self
.set_state(VnfRecordState
.TERMINATE_PENDING
)
1139 with self
._dts
.transaction(flags
=0) as xact
:
1140 block
= xact
.block_create()
1141 block
.add_query_delete(self
.xpath
)
1142 yield from block
.execute(flags
=0)
1143 self
.set_state(VnfRecordState
.TERMINATED
)
1144 self
._log
.debug("Terminated VNF id:%s", self
.id)
1147 class NetworkServiceStatus(object):
1148 """ A class representing the Network service's status """
1149 MAX_EVENTS_RECORDED
= 10
1150 """ Network service Status class"""
1151 def __init__(self
, dts
, log
, loop
):
1156 self
._state
= NetworkServiceRecordState
.INIT
1157 self
._events
= deque([])
1160 def create_notification(self
, evt
, evt_desc
, evt_details
):
1161 xp
= "N,/rw-nsr:nsm-notification"
1162 notif
= RwNsrYang
.YangNotif_RwNsr_NsmNotification()
1164 notif
.description
= evt_desc
1165 notif
.details
= evt_details
if evt_details
is not None else None
1167 yield from self
._dts
.query_create(xp
, rwdts
.XactFlag
.ADVISE
, notif
)
1168 self
._log
.info("Notification called by creating dts query: %s", notif
)
1170 def record_event(self
, evt
, evt_desc
, evt_details
):
1171 """ Record an event """
1172 self
._log
.debug("Recording event - evt %s, evt_descr %s len = %s",
1173 evt
, evt_desc
, len(self
._events
))
1174 if len(self
._events
) >= NetworkServiceStatus
.MAX_EVENTS_RECORDED
:
1175 self
._events
.popleft()
1176 self
._events
.append((int(time
.time()), evt
, evt_desc
,
1177 evt_details
if evt_details
is not None else None))
1179 self
._loop
.create_task(self
.create_notification(evt
,evt_desc
,evt_details
))
1181 def set_state(self
, state
):
1182 """ set the state of this status object """
1186 """ Return the state as a yang enum string """
1187 state_to_str_map
= {"INIT": "init",
1188 "VL_INIT_PHASE": "vl_init_phase",
1189 "VNF_INIT_PHASE": "vnf_init_phase",
1190 "VNFFG_INIT_PHASE": "vnffg_init_phase",
1191 "SCALING_GROUP_INIT_PHASE": "scaling_group_init_phase",
1192 "RUNNING": "running",
1193 "SCALING_OUT": "scaling_out",
1194 "SCALING_IN": "scaling_in",
1195 "TERMINATE_RCVD": "terminate_rcvd",
1196 "TERMINATE": "terminate",
1197 "VL_TERMINATE_PHASE": "vl_terminate_phase",
1198 "VNF_TERMINATE_PHASE": "vnf_terminate_phase",
1199 "VNFFG_TERMINATE_PHASE": "vnffg_terminate_phase",
1200 "TERMINATED": "terminated",
1202 "VL_INSTANTIATE": "vl_instantiate",
1203 "VL_TERMINATE": "vl_terminate",
1205 return state_to_str_map
[self
._state
.name
]
1209 """ State of this status object """
1214 """ Network Service Record as a message"""
1217 for entry
in self
._events
:
1218 event
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_OperationalEvents()
1221 event
.timestamp
, event
.event
, event
.description
, event
.details
= entry
1222 event_list
.append(event
)
1226 class NetworkServiceRecord(object):
1227 """ Network service record """
1228 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
1230 def __init__(self
, dts
, log
, loop
, nsm
, nsm_plugin
, nsr_cfg_msg
, sdn_account_name
, key_pairs
, restart_mode
=False):
1235 self
._nsr
_cfg
_msg
= nsr_cfg_msg
1236 self
._nsm
_plugin
= nsm_plugin
1237 self
._sdn
_account
_name
= sdn_account_name
1240 self
._nsr
_msg
= None
1241 self
._nsr
_regh
= None
1242 self
._key
_pairs
= key_pairs
1247 self
._param
_pools
= {}
1248 self
._scaling
_groups
= {}
1249 self
._create
_time
= int(time
.time())
1250 self
._op
_status
= NetworkServiceStatus(dts
, log
, loop
)
1251 self
._config
_status
= NsrYang
.ConfigStates
.CONFIGURING
1252 self
._config
_status
_details
= None
1254 self
.restart_mode
= restart_mode
1255 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
1256 self
._debug
_running
= False
1257 self
._is
_active
= False
1258 self
._vl
_phase
_completed
= False
1259 self
._vnf
_phase
_completed
= False
1262 # Initalise the state to init
1263 # The NSR moves through the following transitions
1264 # 1. INIT -> VLS_READY once all the VLs in the NSD are created
1265 # 2. VLS_READY - VNFS_READY when all the VNFs in the NSD are created
1266 # 3. VNFS_READY - READY when the NSR is published
1268 self
.set_state(NetworkServiceRecordState
.INIT
)
1270 self
.substitute_input_parameters
= InputParameterSubstitution(self
._log
)
1273 def nsm_plugin(self
):
1275 return self
._nsm
_plugin
1277 def set_state(self
, state
):
1278 """ Set state for this NSR"""
1279 self
._log
.debug("Setting state to %s", state
)
1280 # We are in init phase and is moving to the next state
1281 # The new state could be a FAILED state or VNF_INIIT_PHASE
1282 if self
.state
== NetworkServiceRecordState
.VL_INIT_PHASE
:
1283 self
._vl
_phase
_completed
= True
1285 if self
.state
== NetworkServiceRecordState
.VNF_INIT_PHASE
:
1286 self
._vnf
_phase
_completed
= True
1288 self
._op
_status
.set_state(state
)
1292 """ Get id for this NSR"""
1293 return self
._nsr
_cfg
_msg
.id
1297 """ Name of this network service record """
1298 return self
._nsr
_cfg
_msg
.name
1301 def cloud_account_name(self
):
1302 return self
._nsr
_cfg
_msg
.cloud_account
1305 def om_datacenter_name(self
):
1306 if self
._nsr
_cfg
_msg
.has_field('om_datacenter'):
1307 return self
._nsr
_cfg
_msg
.om_datacenter
1312 """State of this NetworkServiceRecord"""
1313 return self
._op
_status
.state
1317 """ Is this NSR active ?"""
1318 return True if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
else False
1322 """ VLRs associated with this NSR"""
1327 """ VNFRs associated with this NSR"""
1332 """ VNFFGRs associated with this NSR"""
1333 return self
._vnffgrs
1336 def scaling_groups(self
):
1337 """ Scaling groups associated with this NSR """
1338 return self
._scaling
_groups
1341 def param_pools(self
):
1342 """ Parameter value pools associated with this NSR"""
1343 return self
._param
_pools
1346 def nsr_cfg_msg(self
):
1347 return self
._nsr
_cfg
_msg
1350 def nsr_cfg_msg(self
, msg
):
1351 self
._nsr
_cfg
_msg
= msg
1355 """ NSD Protobuf for this NSR """
1356 if self
._nsd
is not None:
1358 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
1363 """ NSD ID for this NSR """
1364 return self
.nsd_msg
.id
1368 ''' Get a new job id for config primitive'''
1373 def config_status(self
):
1374 """ Config status for NSR """
1375 return self
._config
_status
1377 def resolve_placement_group_cloud_construct(self
, input_group
):
1379 Returns the cloud specific construct for placement group
1381 copy_dict
= ['name', 'requirement', 'strategy']
1383 for group_info
in self
._nsr
_cfg
_msg
.nsd_placement_group_maps
:
1384 if group_info
.placement_group_ref
== input_group
.name
:
1385 group
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_PlacementGroupsInfo()
1386 group_dict
= {k
:v
for k
,v
in
1387 group_info
.as_dict().items() if k
!= 'placement_group_ref'}
1388 for param
in copy_dict
:
1389 group_dict
.update({param
: getattr(input_group
, param
)})
1390 group
.from_dict(group_dict
)
1396 return "NSR(name={}, nsd_id={}, cloud_account={})".format(
1397 self
.name
, self
.nsd_id
, self
.cloud_account_name
1400 def _get_vnfd(self
, vnfd_id
, config_xact
):
1401 """ Fetch vnfd msg for the passed vnfd id """
1402 return self
._nsm
.get_vnfd(vnfd_id
, config_xact
)
1404 def _get_vnfd_cloud_account(self
, vnfd_member_index
):
1405 """ Fetch Cloud Account for the passed vnfd id """
1406 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1407 vim_accounts
= [(vnf
.cloud_account
,vnf
.om_datacenter
) for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map \
1408 if vnfd_member_index
== vnf
.member_vnf_index_ref
]
1409 if vim_accounts
and vim_accounts
[0]:
1410 return vim_accounts
[0]
1411 return (self
.cloud_account_name
,self
.om_datacenter_name
)
1413 def _get_constituent_vnfd_msg(self
, vnf_index
):
1414 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1415 if const_vnfd
.member_vnf_index
== vnf_index
:
1418 raise ValueError("Constituent VNF index %s not found" % vnf_index
)
1420 def record_event(self
, evt
, evt_desc
, evt_details
=None, state
=None):
1421 """ Record an event """
1422 self
._op
_status
.record_event(evt
, evt_desc
, evt_details
)
1423 if state
is not None:
1424 self
.set_state(state
)
1426 def scaling_trigger_str(self
, trigger
):
1427 SCALING_TRIGGER_STRS
= {
1428 NsdYang
.ScalingTrigger
.PRE_SCALE_IN
: 'pre-scale-in',
1429 NsdYang
.ScalingTrigger
.POST_SCALE_IN
: 'post-scale-in',
1430 NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
: 'pre-scale-out',
1431 NsdYang
.ScalingTrigger
.POST_SCALE_OUT
: 'post-scale-out',
1434 return SCALING_TRIGGER_STRS
[trigger
]
1435 except Exception as e
:
1436 self
._log
.error("Scaling trigger mapping error for {} : {}".
1438 self
._log
.exception(e
)
1439 return "Unknown trigger"
1442 def instantiate_vls(self
):
1444 This function instantiates VLs for every VL in this Network Service
1446 self
._log
.debug("Instantiating %d VLs in NSD id %s", len(self
._vlrs
),
1448 for vlr
in self
._vlrs
:
1449 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1450 vlr
.state
= VlRecordState
.ACTIVE
1453 def create(self
, config_xact
):
1454 """ Create this network service"""
1455 # Create virtual links for all the external vnf
1456 # connection points in this NS
1457 yield from self
.create_vls()
1459 # Create VNFs in this network service
1460 yield from self
.create_vnfs(config_xact
)
1462 # Create VNFFG for network service
1463 self
.create_vnffgs()
1465 # Create Scaling Groups for each scaling group in NSD
1466 self
.create_scaling_groups()
1468 # Create Parameter Pools
1469 self
.create_param_pools()
1472 def apply_scale_group_config_script(self
, script
, group
, scale_instance
, trigger
, vnfrs
=None):
1473 """ Apply config based on script for scale group """
1476 def add_vnfrs_data(vnfrs_list
):
1477 """ Add as a dict each of the VNFRs data """
1479 for vnfr
in vnfrs_list
:
1480 self
._log
.debug("Add VNFR {} data".format(vnfr
))
1482 vnfr_data
['name'] = vnfr
.name
1483 if trigger
in [NsdYang
.ScalingTrigger
.PRE_SCALE_IN
, NsdYang
.ScalingTrigger
.POST_SCALE_OUT
]:
1484 # Get VNF management and other IPs, etc
1485 opdata
= yield from self
.fetch_vnfr(vnfr
.xpath
)
1486 self
._log
.debug("VNFR {} op data: {}".format(vnfr
.name
, opdata
))
1488 vnfr_data
['rw_mgmt_ip'] = opdata
.mgmt_interface
.ip_address
1489 vnfr_data
['rw_mgmt_port'] = opdata
.mgmt_interface
.port
1490 except Exception as e
:
1491 self
._log
.error("Unable to get management IP for vnfr {}:{}".
1492 format(vnfr
.name
, e
))
1495 vnfr_data
['connection_points'] = []
1496 for cp
in opdata
.connection_point
:
1498 con_pt
['name'] = cp
.name
1499 con_pt
['ip_address'] = cp
.ip_address
1500 vnfr_data
['connection_points'].append(con_pt
)
1501 except Exception as e
:
1502 self
._log
.error("Exception getting connections points for VNFR {}: {}".
1503 format(vnfr
.name
, e
))
1505 vnfrs_data
.append(vnfr_data
)
1506 self
._log
.debug("VNFRs data: {}".format(vnfrs_data
))
1510 def add_nsr_data(nsr
):
1512 nsr_data
['name'] = nsr
.name
1515 if script
is None or len(script
) == 0:
1516 self
._log
.error("Script not provided for scale group config: {}".format(group
.name
))
1519 if script
[0] == '/':
1522 path
= os
.path
.join(os
.environ
['RIFT_INSTALL'], "usr/bin", script
)
1523 if not os
.path
.exists(path
):
1524 self
._log
.error("Config faled for scale group {}: Script does not exist at {}".
1525 format(group
.name
, path
))
1528 # Build a YAML file with all parameters for the script to execute
1529 # The data consists of 5 sections
1531 # 2. Scale group config
1532 # 3. VNFRs in the scale group
1533 # 4. VNFRs outside scale group
1536 data
['trigger'] = group
.trigger_map(trigger
)
1537 data
['config'] = group
.group_msg
.as_dict()
1540 data
["vnfrs_in_group"] = yield from add_vnfrs_data(vnfrs
)
1542 data
["vnfrs_in_group"] = yield from add_vnfrs_data(scale_instance
.vnfrs
)
1544 data
["vnfrs_others"] = yield from add_vnfrs_data(self
.vnfrs
.values())
1545 data
["nsr"] = add_nsr_data(self
)
1548 with tempfile
.NamedTemporaryFile(delete
=False) as tmp_file
:
1549 tmp_file
.write(yaml
.dump(data
, default_flow_style
=True)
1552 self
._log
.debug("Creating a temp file: {} with input data: {}".
1553 format(tmp_file
.name
, data
))
1555 cmd
= "{} {}".format(path
, tmp_file
.name
)
1556 self
._log
.debug("Running the CMD: {}".format(cmd
))
1557 proc
= yield from asyncio
.create_subprocess_shell(cmd
, loop
=self
._loop
)
1558 rc
= yield from proc
.wait()
1560 self
._log
.error("The script {} for scale group {} config returned: {}".
1561 format(script
, group
.name
, rc
))
1569 def apply_scaling_group_config(self
, trigger
, group
, scale_instance
, vnfrs
=None):
1570 """ Apply the config for the scaling group based on trigger """
1571 if group
is None or scale_instance
is None:
1575 def update_config_status(success
=True, err_msg
=None):
1576 self
._log
.debug("Update %s config status to %r : %s",
1577 scale_instance
, success
, err_msg
)
1578 if (scale_instance
.config_status
== "failed"):
1579 # Do not update the config status if it is already in failed state
1582 if scale_instance
.config_status
== "configured":
1583 # Update only to failed state an already configured scale instance
1585 scale_instance
.config_status
= "failed"
1586 scale_instance
.config_err_msg
= err_msg
1587 yield from self
.update_state()
1589 # We are in configuring state
1590 # Only after post scale out mark instance as configured
1591 if trigger
== NsdYang
.ScalingTrigger
.POST_SCALE_OUT
:
1593 scale_instance
.config_status
= "configured"
1595 scale_instance
.config_status
= "failed"
1596 scale_instance
.config_err_msg
= err_msg
1597 yield from self
.update_state()
1599 config
= group
.trigger_config(trigger
)
1603 self
._log
.debug("Scaling group {} config: {}".format(group
.name
, config
))
1604 if config
.has_field("ns_config_primitive_name_ref"):
1605 config_name
= config
.ns_config_primitive_name_ref
1606 nsd_msg
= self
.nsd_msg
1607 config_primitive
= None
1608 for ns_cfg_prim
in nsd_msg
.service_primitive
:
1609 if ns_cfg_prim
.name
== config_name
:
1610 config_primitive
= ns_cfg_prim
1613 if config_primitive
is None:
1614 raise ValueError("Could not find ns_cfg_prim %s in nsr %s" % (config_name
, self
.name
))
1616 self
._log
.debug("Scaling group {} config primitive: {}".format(group
.name
, config_primitive
))
1617 if config_primitive
.has_field("user_defined_script"):
1618 rc
= yield from self
.apply_scale_group_config_script(config_primitive
.user_defined_script
,
1619 group
, scale_instance
, trigger
, vnfrs
)
1622 err_msg
= "Failed config for trigger {} using config script '{}'". \
1623 format(self
.scaling_trigger_str(trigger
),
1624 config_primitive
.user_defined_script
)
1625 yield from update_config_status(success
=rc
, err_msg
=err_msg
)
1628 err_msg
= "Failed config for trigger {} as config script is not specified". \
1629 format(self
.scaling_trigger_str(trigger
))
1630 yield from update_config_status(success
=False, err_msg
=err_msg
)
1631 raise NotImplementedError("Only script based config support for scale group for now: {}".
1634 err_msg
= "Failed config for trigger {} as config primitive is not specified".\
1635 format(self
.scaling_trigger_str(trigger
))
1636 yield from update_config_status(success
=False, err_msg
=err_msg
)
1637 self
._log
.error("Config primitive not specified for config action in scale group %s" %
1641 def create_scaling_groups(self
):
1642 """ This function creates a NSScalingGroup for every scaling
1643 group defined in he NSD"""
1645 for scaling_group_msg
in self
.nsd_msg
.scaling_group_descriptor
:
1646 self
._log
.debug("Found scaling_group %s in nsr id %s",
1647 scaling_group_msg
.name
, self
.id)
1649 group_record
= scale_group
.ScalingGroup(
1654 self
._scaling
_groups
[group_record
.name
] = group_record
1657 def create_scale_group_instance(self
, group_name
, index
, config_xact
, is_default
=False):
1658 group
= self
._scaling
_groups
[group_name
]
1659 scale_instance
= group
.create_instance(index
, is_default
)
1663 self
._log
.debug("Creating %u VNFs associated with NS id %s scaling group %s",
1664 len(self
.nsd_msg
.constituent_vnfd
), self
.id, self
)
1667 for vnf_index
, count
in group
.vnf_index_count_map
.items():
1668 const_vnfd_msg
= self
._get
_constituent
_vnfd
_msg
(vnf_index
)
1669 vnfd_msg
= self
._get
_vnfd
(const_vnfd_msg
.vnfd_id_ref
, config_xact
)
1671 cloud_account_name
, om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd_msg
.member_vnf_index
)
1672 if cloud_account_name
is None:
1673 cloud_account_name
= self
.cloud_account_name
1674 for _
in range(count
):
1675 vnfr
= yield from self
.create_vnf_record(vnfd_msg
, const_vnfd_msg
, cloud_account_name
, om_datacenter_name
, group_name
, index
)
1676 scale_instance
.add_vnfr(vnfr
)
1682 def instantiate_instance():
1683 self
._log
.debug("Creating %s VNFRS", scale_instance
)
1684 vnfrs
= yield from create_vnfs()
1685 yield from self
.publish()
1687 self
._log
.debug("Instantiating %s VNFRS for %s", len(vnfrs
), scale_instance
)
1688 scale_instance
.operational_status
= "vnf_init_phase"
1689 yield from self
.update_state()
1692 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
,
1693 group
, scale_instance
, vnfrs
)
1695 self
._log
.error("Pre scale out config for scale group {} ({}) failed".
1696 format(group
.name
, index
))
1697 scale_instance
.operational_status
= "failed"
1699 yield from self
.instantiate_vnfs(vnfrs
)
1701 except Exception as e
:
1702 self
._log
.exception("Failed to begin instantiatiation of vnfs for scale group {}: {}".
1703 format(group
.name
, e
))
1704 self
._log
.exception(e
)
1705 scale_instance
.operational_status
= "failed"
1707 yield from self
.update_state()
1709 yield from instantiate_instance()
1712 def delete_scale_group_instance(self
, group_name
, index
):
1713 group
= self
._scaling
_groups
[group_name
]
1714 scale_instance
= group
.get_instance(index
)
1715 if scale_instance
.is_default
:
1716 raise ScalingOperationError("Cannot terminate a default scaling group instance")
1718 scale_instance
.operational_status
= "terminate"
1719 yield from self
.update_state()
1722 def terminate_instance():
1723 self
._log
.debug("Terminating %s VNFRS" % scale_instance
)
1724 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_IN
,
1725 group
, scale_instance
)
1727 self
._log
.error("Pre scale in config for scale group {} ({}) failed".
1728 format(group
.name
, index
))
1730 # Going ahead with terminate, even if there is an error in pre-scale-in config
1731 # as this could be result of scale out failure and we need to cleanup this group
1732 yield from self
.terminate_vnfrs(scale_instance
.vnfrs
)
1733 group
.delete_instance(index
)
1735 scale_instance
.operational_status
= "vnf_terminate_phase"
1736 yield from self
.update_state()
1738 yield from terminate_instance()
1741 def _update_scale_group_instances_status(self
):
1743 def post_scale_out_task(group
, instance
):
1744 # Apply post scale out config once all VNFRs are active
1745 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_OUT
,
1747 instance
.operational_status
= "running"
1749 self
._log
.debug("Scale out for group {} and instance {} succeeded".
1750 format(group
.name
, instance
.instance_id
))
1752 self
._log
.error("Post scale out config for scale group {} ({}) failed".
1753 format(group
.name
, instance
.instance_id
))
1755 yield from self
.update_state()
1757 group_instances
= {group
: group
.instances
for group
in self
._scaling
_groups
.values()}
1758 for group
, instances
in group_instances
.items():
1759 self
._log
.debug("Updating %s instance status", group
)
1760 for instance
in instances
:
1761 instance_vnf_state_list
= [vnfr
.state
for vnfr
in instance
.vnfrs
]
1762 self
._log
.debug("Got vnfr instance states: %s", instance_vnf_state_list
)
1763 if instance
.operational_status
== "vnf_init_phase":
1764 if all([state
== VnfRecordState
.ACTIVE
for state
in instance_vnf_state_list
]):
1765 instance
.operational_status
= "running"
1767 # Create a task for post scale out to allow us to sleep before attempting
1768 # to configure newly created VM's
1769 self
._loop
.create_task(post_scale_out_task(group
, instance
))
1771 elif any([state
== VnfRecordState
.FAILED
for state
in instance_vnf_state_list
]):
1772 self
._log
.debug("Scale out for group {} and instance {} failed".
1773 format(group
.name
, instance
.instance_id
))
1774 instance
.operational_status
= "failed"
1776 elif instance
.operational_status
== "vnf_terminate_phase":
1777 if all([state
== VnfRecordState
.TERMINATED
for state
in instance_vnf_state_list
]):
1778 instance
.operational_status
= "terminated"
1779 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_IN
,
1782 self
._log
.debug("Scale in for group {} and instance {} succeeded".
1783 format(group
.name
, instance
.instance_id
))
1785 self
._log
.error("Post scale in config for scale group {} ({}) failed".
1786 format(group
.name
, instance
.instance_id
))
1788 def create_vnffgs(self
):
1789 """ This function creates VNFFGs for every VNFFG in the NSD
1790 associated with this NSR"""
1792 for vnffgd
in self
.nsd_msg
.vnffgd
:
1793 self
._log
.debug("Found vnffgd %s in nsr id %s", vnffgd
, self
.id)
1794 vnffgr
= VnffgRecord(self
._dts
,
1797 self
._nsm
._vnffgmgr
,
1801 self
._sdn
_account
_name
1803 self
._vnffgrs
[vnffgr
.id] = vnffgr
1805 def resolve_vld_ip_profile(self
, nsd_msg
, vld
):
1806 self
._log
.debug("Receieved ip profile ref is %s",vld
.ip_profile_ref
)
1807 if not vld
.has_field('ip_profile_ref'):
1809 profile
= [profile
for profile
in nsd_msg
.ip_profiles
if profile
.name
== vld
.ip_profile_ref
]
1810 return profile
[0] if profile
else None
1813 def _create_vls(self
, vld
, cloud_account
,om_datacenter
):
1814 """Create a VLR in the cloud account specified using the given VLD
1818 cloud_account : Cloud account name
1823 vlr
= yield from VirtualLinkRecord
.create_record(
1831 self
.resolve_vld_ip_profile(self
.nsd_msg
, vld
),
1833 restart_mode
=self
.restart_mode
)
1837 def _extract_cloud_accounts_for_vl(self
, vld
):
1839 Extracts the list of cloud accounts from the NS Config obj
1842 1. Cloud accounts based connection point (vnf_cloud_account_map)
1844 vld : VLD yang object
1849 cloud_account_list
= []
1851 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1852 # Handle case where cloud_account is None
1854 for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1855 if vnf
.cloud_account
is not None or vnf
.om_datacenter
is not None:
1856 vnf_cloud_map
[vnf
.member_vnf_index_ref
] = (vnf
.cloud_account
,vnf
.om_datacenter
)
1858 for vnfc
in vld
.vnfd_connection_point_ref
:
1859 cloud_account
= vnf_cloud_map
.get(
1860 vnfc
.member_vnf_index_ref
,
1861 (self
.cloud_account_name
,self
.om_datacenter_name
))
1863 cloud_account_list
.append(cloud_account
)
1865 if self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1866 for vld_map
in self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1867 if vld_map
.vld_id_ref
== vld
.id:
1868 for cloud_account
in vld_map
.cloud_accounts
:
1869 cloud_account_list
.extend((cloud_account
,None))
1870 for om_datacenter
in vld_map
.om_datacenters
:
1871 cloud_account_list
.extend((None,om_datacenter
))
1873 # If no config has been provided then fall-back to the default
1875 if not cloud_account_list
:
1876 cloud_account_list
= [(self
.cloud_account_name
,self
.om_datacenter_name
)]
1878 self
._log
.debug("VL {} cloud accounts: {}".
1879 format(vld
.name
, cloud_account_list
))
1880 return set(cloud_account_list
)
1883 def create_vls(self
):
1884 """ This function creates VLs for every VLD in the NSD
1885 associated with this NSR"""
1886 for vld
in self
.nsd_msg
.vld
:
1887 self
._log
.debug("Found vld %s in nsr id %s", vld
, self
.id)
1888 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1889 for cloud_account
,om_datacenter
in cloud_account_list
:
1890 vlr
= yield from self
._create
_vls
(vld
, cloud_account
,om_datacenter
)
1891 self
._vlrs
.append(vlr
)
1895 def create_vl_instance(self
, vld
):
1896 self
._log
.debug("Create VL for {}: {}".format(self
.id, vld
.as_dict()))
1897 # Check if the VL is already present
1899 for vl
in self
._vlrs
:
1900 if vl
.vld_msg
.id == vld
.id:
1901 self
._log
.debug("The VLD %s already in NSR %s as VLR %s with status %s",
1902 vld
.id, self
.id, vl
.id, vl
.state
)
1904 if vlr
.state
!= VlRecordState
.TERMINATED
:
1905 err_msg
= "VLR for VL %s in NSR %s already instantiated", \
1907 self
._log
.error(err_msg
)
1908 raise NsrVlUpdateError(err_msg
)
1912 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1913 for account
,om_datacenter
in cloud_account_list
:
1914 vlr
= yield from self
._create
_vls
(vld
, account
,om_datacenter
)
1915 self
._vlrs
.append(vlr
)
1917 vlr
.state
= VlRecordState
.INSTANTIATION_PENDING
1918 yield from self
.update_state()
1921 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1922 vlr
.state
= VlRecordState
.ACTIVE
1924 except Exception as e
:
1925 err_msg
= "Error instantiating VL for NSR {} and VLD {}: {}". \
1926 format(self
.id, vld
.id, e
)
1927 self
._log
.error(err_msg
)
1928 self
._log
.exception(e
)
1929 vlr
.state
= VlRecordState
.FAILED
1931 yield from self
.update_state()
1934 def delete_vl_instance(self
, vld
):
1935 for vlr
in self
._vlrs
:
1936 if vlr
.vld_msg
.id == vld
.id:
1937 self
._log
.debug("Found VLR %s for VLD %s in NSR %s",
1938 vlr
.id, vld
.id, self
.id)
1939 vlr
.state
= VlRecordState
.TERMINATE_PENDING
1940 yield from self
.update_state()
1943 yield from self
.nsm_plugin
.terminate_vl(vlr
)
1944 vlr
.state
= VlRecordState
.TERMINATED
1945 self
._vlrs
.remove(vlr
)
1947 except Exception as e
:
1948 err_msg
= "Error terminating VL for NSR {} and VLD {}: {}". \
1949 format(self
.id, vld
.id, e
)
1950 self
._log
.error(err_msg
)
1951 self
._log
.exception(e
)
1952 vlr
.state
= VlRecordState
.FAILED
1954 yield from self
.update_state()
1958 def create_vnfs(self
, config_xact
):
1960 This function creates VNFs for every VNF in the NSD
1961 associated with this NSR
1963 self
._log
.debug("Creating %u VNFs associated with this NS id %s",
1964 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
1966 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1967 if not const_vnfd
.start_by_default
:
1968 self
._log
.debug("start_by_default set to False in constituent VNF (%s). Skipping start.",
1969 const_vnfd
.member_vnf_index
)
1972 vnfd_msg
= self
._get
_vnfd
(const_vnfd
.vnfd_id_ref
, config_xact
)
1973 cloud_account_name
,om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd
.member_vnf_index
)
1974 if cloud_account_name
is None:
1975 cloud_account_name
= self
.cloud_account_name
1976 yield from self
.create_vnf_record(vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
)
1979 def get_placement_groups(self
, vnfd_msg
, const_vnfd
):
1980 placement_groups
= []
1981 for group
in self
.nsd_msg
.placement_groups
:
1982 for member_vnfd
in group
.member_vnfd
:
1983 if (member_vnfd
.vnfd_id_ref
== vnfd_msg
.id) and \
1984 (member_vnfd
.member_vnf_index_ref
== const_vnfd
.member_vnf_index
):
1985 group_info
= self
.resolve_placement_group_cloud_construct(group
)
1986 if group_info
is None:
1987 self
._log
.error("Could not resolve cloud-construct for placement group: %s", group
.name
)
1988 ### raise PlacementGroupError("Could not resolve cloud-construct for placement group: {}".format(group.name))
1990 self
._log
.info("Successfully resolved cloud construct for placement group: %s for VNF: %s (Member Index: %s)",
1993 const_vnfd
.member_vnf_index
)
1994 placement_groups
.append(group_info
)
1995 return placement_groups
1998 def create_vnf_record(self
, vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
, group_name
=None, group_instance_id
=None):
1999 # Fetch the VNFD associated with this VNF
2000 placement_groups
= self
.get_placement_groups(vnfd_msg
, const_vnfd
)
2001 self
._log
.info("Cloud Account for VNF %d is %s",const_vnfd
.member_vnf_index
,cloud_account_name
)
2002 self
._log
.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s",
2004 const_vnfd
.member_vnf_index
,
2005 [ group
.name
for group
in placement_groups
])
2006 vnfr
= yield from VirtualNetworkFunctionRecord
.create_record(self
._dts
,
2019 restart_mode
=self
.restart_mode
,
2021 if vnfr
.id in self
._vnfrs
:
2022 err
= "VNF with VNFR id %s already in vnf list" % (vnfr
.id,)
2023 raise NetworkServiceRecordError(err
)
2025 self
._vnfrs
[vnfr
.id] = vnfr
2026 self
._nsm
.vnfrs
[vnfr
.id] = vnfr
2028 yield from vnfr
.set_config_status(NsrYang
.ConfigStates
.INIT
)
2030 self
._log
.debug("Added VNFR %s to NSM VNFR list with id %s",
2036 def create_param_pools(self
):
2037 for param_pool
in self
.nsd_msg
.parameter_pool
:
2038 self
._log
.debug("Found parameter pool %s in nsr id %s", param_pool
, self
.id)
2040 start_value
= param_pool
.range.start_value
2041 end_value
= param_pool
.range.end_value
2042 if end_value
< start_value
:
2043 raise NetworkServiceRecordError(
2044 "Parameter pool %s has invalid range (start: {}, end: {})".format(
2045 start_value
, end_value
2049 self
._param
_pools
[param_pool
.name
] = config_value_pool
.ParameterValuePool(
2052 range(start_value
, end_value
)
2056 def fetch_vnfr(self
, vnfr_path
):
2057 """ Fetch VNFR record """
2059 self
._log
.debug("Fetching VNFR with key %s while instantiating %s",
2061 res_iter
= yield from self
._dts
.query_read(vnfr_path
, rwdts
.XactFlag
.MERGE
)
2063 for ent
in res_iter
:
2064 res
= yield from ent
2070 def instantiate_vnfs(self
, vnfrs
):
2072 This function instantiates VNFs for every VNF in this Network Service
2074 self
._log
.debug("Instantiating %u VNFs in NS %s", len(vnfrs
), self
.id)
2076 self
._log
.debug("Instantiating VNF: %s in NS %s", vnf
, self
.id)
2077 yield from self
.nsm_plugin
.instantiate_vnf(self
, vnf
)
2080 def instantiate_vnffgs(self
):
2082 This function instantiates VNFFGs for every VNFFG in this Network Service
2084 self
._log
.debug("Instantiating %u VNFFGs in NS %s",
2085 len(self
.nsd_msg
.vnffgd
), self
.id)
2086 for _
, vnfr
in self
.vnfrs
.items():
2087 while vnfr
.state
in [VnfRecordState
.INSTANTIATION_PENDING
, VnfRecordState
.INIT
]:
2088 self
._log
.debug("Received vnfr state for vnfr %s is %s; retrying",vnfr
.name
,vnfr
.state
)
2089 yield from asyncio
.sleep(2, loop
=self
._loop
)
2090 if vnfr
.state
== VnfRecordState
.ACTIVE
:
2091 self
._log
.debug("Received vnfr state for vnfr %s is %s ",vnfr
.name
,vnfr
.state
)
2094 self
._log
.debug("Received vnfr state for vnfr %s is %s; failing vnffg creation",vnfr
.name
,vnfr
.state
)
2095 self
._vnffgr
_state
= VnffgRecordState
.FAILED
2098 self
._log
.info("Waiting for 90 seconds for VMs to come up")
2099 yield from asyncio
.sleep(90, loop
=self
._loop
)
2100 self
._log
.info("Starting VNFFG orchestration")
2101 for vnffg
in self
._vnffgrs
.values():
2102 self
._log
.debug("Instantiating VNFFG: %s in NS %s", vnffg
, self
.id)
2103 yield from vnffg
.instantiate()
2106 def instantiate_scaling_instances(self
, config_xact
):
2107 """ Instantiate any default scaling instances in this Network Service """
2108 for group
in self
._scaling
_groups
.values():
2109 for i
in range(group
.min_instance_count
):
2110 self
._log
.debug("Instantiating %s default scaling instance %s", group
, i
)
2111 yield from self
.create_scale_group_instance(
2112 group
.name
, i
, config_xact
, is_default
=True
2115 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2116 if group_msg
.scaling_group_name_ref
!= group
.name
:
2119 for instance
in group_msg
.instance
:
2120 self
._log
.debug("Reloading %s scaling instance %s", group_msg
, instance
.id)
2121 yield from self
.create_scale_group_instance(
2122 group
.name
, instance
.id, config_xact
, is_default
=False
2125 def has_scaling_instances(self
):
2126 """ Return boolean indicating if the network service has default scaling groups """
2127 for group
in self
._scaling
_groups
.values():
2128 if group
.min_instance_count
> 0:
2131 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2132 if len(group_msg
.instance
) > 0:
2139 """ This function publishes this NSR """
2140 self
._nsr
_msg
= self
.create_msg()
2142 self
._log
.debug("Publishing the NSR with xpath %s and nsr %s",
2146 if self
._debug
_running
:
2147 self
._log
.debug("Publishing NSR in RUNNING state!")
2150 with self
._dts
.transaction() as xact
:
2151 yield from self
._nsm
.nsr_handler
.update(xact
, self
.nsr_xpath
, self
._nsr
_msg
)
2152 if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
:
2153 self
._debug
_running
= True
2156 def unpublish(self
, xact
):
2157 """ Unpublish this NSR object """
2158 self
._log
.debug("Unpublishing Network service id %s", self
.id)
2159 yield from self
._nsm
.nsr_handler
.delete(xact
, self
.nsr_xpath
)
2162 def nsr_xpath(self
):
2163 """ Returns the xpath associated with this NSR """
2165 "D,/nsr:ns-instance-opdata" +
2166 "/nsr:nsr[nsr:ns-instance-config-ref = '{}']"
2170 def xpath_from_nsr(nsr
):
2171 """ Returns the xpath associated with this NSR op data"""
2172 return (NetworkServiceRecord
.XPATH
+
2173 "[nsr:ns-instance-config-ref = '{}']").format(nsr
.id)
2176 def nsd_xpath(self
):
2177 """ Return NSD config xpath."""
2179 "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}']"
2180 ).format(self
.nsd_id
)
2183 def instantiate(self
, config_xact
):
2184 """"Instantiates a NetworkServiceRecord.
2186 This function instantiates a Network service
2187 which involves the following steps,
2189 * Instantiate every VL in NSD by sending create VLR request to DTS.
2190 * Instantiate every VNF in NSD by sending create VNF reuqest to DTS.
2191 * Publish the NSR details to DTS
2194 nsr: The NSR configuration request containing nsr-id and nsd
2195 config_xact: The configuration transaction which initiated the instatiation
2198 NetworkServiceRecordError if the NSR creation fails
2204 self
._log
.debug("Instantiating NS - %s xact - %s", self
, config_xact
)
2206 # Move the state to INIITALIZING
2207 self
.set_state(NetworkServiceRecordState
.INIT
)
2209 event_descr
= "Instantiation Request Received NSR Id:%s" % self
.id
2210 self
.record_event("instantiating", event_descr
)
2213 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
2216 # Update ref count if nsd present in catalog
2217 self
._nsm
.get_nsd_ref(self
.nsd_id
)
2219 except NetworkServiceDescriptorError
:
2220 # This could be an NSD not in the nsd-catalog
2223 # Merge any config and initial config primitive values
2224 self
.config_store
.merge_nsd_config(self
.nsd_msg
)
2225 self
._log
.debug("Merged NSD: {}".format(self
.nsd_msg
.as_dict()))
2227 event_descr
= "Fetched NSD with descriptor id %s" % self
.nsd_id
2228 self
.record_event("nsd-fetched", event_descr
)
2230 if self
._nsd
is None:
2231 msg
= "Failed to fetch NSD with nsd-id [%s] for nsr-id %s"
2232 self
._log
.debug(msg
, self
.nsd_id
, self
.id)
2233 raise NetworkServiceRecordError(self
)
2235 self
._log
.debug("Got nsd result %s", self
._nsd
)
2237 # Substitute any input parameters
2238 self
.substitute_input_parameters(self
._nsd
, self
._nsr
_cfg
_msg
)
2241 yield from self
.create(config_xact
)
2243 # Publish the NSR to DTS
2244 yield from self
.publish()
2247 def do_instantiate():
2249 Instantiate network service
2251 self
._log
.debug("Instantiating VLs nsr id [%s] nsd id [%s]",
2252 self
.id, self
.nsd_id
)
2254 # instantiate the VLs
2255 event_descr
= ("Instantiating %s external VLs for NSR id %s" %
2256 (len(self
.nsd_msg
.vld
), self
.id))
2257 self
.record_event("begin-external-vls-instantiation", event_descr
)
2259 self
.set_state(NetworkServiceRecordState
.VL_INIT_PHASE
)
2261 yield from self
.instantiate_vls()
2263 # Publish the NSR to DTS
2264 yield from self
.publish()
2266 event_descr
= ("Finished instantiating %s external VLs for NSR id %s" %
2267 (len(self
.nsd_msg
.vld
), self
.id))
2268 self
.record_event("end-external-vls-instantiation", event_descr
)
2270 self
.set_state(NetworkServiceRecordState
.VNF_INIT_PHASE
)
2272 self
._log
.debug("Instantiating VNFs ...... nsr[%s], nsd[%s]",
2273 self
.id, self
.nsd_id
)
2275 # instantiate the VNFs
2276 event_descr
= ("Instantiating %s VNFS for NSR id %s" %
2277 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2279 self
.record_event("begin-vnf-instantiation", event_descr
)
2281 yield from self
.instantiate_vnfs(self
._vnfrs
.values())
2283 self
._log
.debug(" Finished instantiating %d VNFs for NSR id %s",
2284 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
2286 event_descr
= ("Finished instantiating %s VNFs for NSR id %s" %
2287 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2288 self
.record_event("end-vnf-instantiation", event_descr
)
2290 if len(self
.vnffgrs
) > 0:
2291 #self.set_state(NetworkServiceRecordState.VNFFG_INIT_PHASE)
2292 event_descr
= ("Instantiating %s VNFFGS for NSR id %s" %
2293 (len(self
.nsd_msg
.vnffgd
), self
.id))
2295 self
.record_event("begin-vnffg-instantiation", event_descr
)
2297 yield from self
.instantiate_vnffgs()
2299 event_descr
= ("Finished instantiating %s VNFFGDs for NSR id %s" %
2300 (len(self
.nsd_msg
.vnffgd
), self
.id))
2301 self
.record_event("end-vnffg-instantiation", event_descr
)
2303 if self
.has_scaling_instances():
2304 event_descr
= ("Instantiating %s Scaling Groups for NSR id %s" %
2305 (len(self
._scaling
_groups
), self
.id))
2307 self
.record_event("begin-scaling-group-instantiation", event_descr
)
2308 yield from self
.instantiate_scaling_instances(config_xact
)
2309 self
.record_event("end-scaling-group-instantiation", event_descr
)
2311 # Give the plugin a chance to deploy the network service now that all
2312 # virtual links and vnfs are instantiated
2313 yield from self
.nsm_plugin
.deploy(self
._nsr
_msg
)
2315 self
._log
.debug("Publishing NSR...... nsr[%s], nsd[%s]",
2316 self
.id, self
.nsd_id
)
2318 # Publish the NSR to DTS
2319 yield from self
.publish()
2321 self
._log
.debug("Published NSR...... nsr[%s], nsd[%s]",
2322 self
.id, self
.nsd_id
)
2324 def on_instantiate_done(fut
):
2325 # If the do_instantiate fails, then publish NSR with failed result
2326 if fut
.exception() is not None:
2327 self
._log
.error("NSR instantiation failed for NSR id %s: %s", self
.id, str(fut
.exception()))
2328 self
._loop
.create_task(self
.instantiation_failed(failed_reason
=str(fut
.exception())))
2330 instantiate_task
= self
._loop
.create_task(do_instantiate())
2331 instantiate_task
.add_done_callback(on_instantiate_done
)
2334 def set_config_status(self
, status
, status_details
=None):
2335 if self
.config_status
!= status
:
2336 self
._log
.debug("Updating NSR {} status for {} to {}".
2337 format(self
.name
, self
.config_status
, status
))
2338 self
._config
_status
= status
2339 self
._config
_status
_details
= status_details
2341 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2342 self
.record_event("config-failed", "NS configuration failed",
2343 evt_details
=self
._config
_status
_details
)
2345 yield from self
.publish()
2348 def is_active(self
):
2349 """ This NS is active """
2350 self
.set_state(NetworkServiceRecordState
.RUNNING
)
2354 # Publish the NSR to DTS
2355 self
._log
.debug("Network service %s is active ", self
.id)
2356 self
._is
_active
= True
2358 event_descr
= "NSR in running state for NSR id %s" % self
.id
2359 self
.record_event("ns-running", event_descr
)
2361 yield from self
.publish()
2364 def instantiation_failed(self
, failed_reason
=None):
2365 """ The NS instantiation failed"""
2366 self
._log
.error("Network service id:%s, name:%s instantiation failed",
2368 self
.set_state(NetworkServiceRecordState
.FAILED
)
2370 event_descr
= "Instantiation of NS %s failed" % self
.id
2371 self
.record_event("ns-failed", event_descr
, evt_details
=failed_reason
)
2373 # Publish the NSR to DTS
2374 yield from self
.publish()
2377 def terminate_vnfrs(self
, vnfrs
):
2378 """ Terminate VNFRS in this network service """
2379 self
._log
.debug("Terminating VNFs in network service %s", self
.id)
2381 yield from self
.nsm_plugin
.terminate_vnf(vnfr
)
2384 def terminate(self
):
2385 """ Terminate a NetworkServiceRecord."""
2386 def terminate_vnffgrs():
2387 """ Terminate VNFFGRS in this network service """
2388 self
._log
.debug("Terminating VNFFGRs in network service %s", self
.id)
2389 for vnffgr
in self
.vnffgrs
.values():
2390 yield from vnffgr
.terminate()
2392 def terminate_vlrs():
2393 """ Terminate VLRs in this netork service """
2394 self
._log
.debug("Terminating VLs in network service %s", self
.id)
2395 for vlr
in self
.vlrs
:
2396 yield from self
.nsm_plugin
.terminate_vl(vlr
)
2397 vlr
.state
= VlRecordState
.TERMINATED
2399 self
._log
.debug("Terminating network service id %s", self
.id)
2401 # Move the state to TERMINATE
2402 self
.set_state(NetworkServiceRecordState
.TERMINATE
)
2403 event_descr
= "Terminate being processed for NS Id:%s" % self
.id
2404 self
.record_event("terminate", event_descr
)
2406 # Move the state to VNF_TERMINATE_PHASE
2407 self
._log
.debug("Terminating VNFFGs in NS ID: %s", self
.id)
2408 self
.set_state(NetworkServiceRecordState
.VNFFG_TERMINATE_PHASE
)
2409 event_descr
= "Terminating VNFFGS in NS Id:%s" % self
.id
2410 self
.record_event("terminating-vnffgss", event_descr
)
2411 yield from terminate_vnffgrs()
2413 # Move the state to VNF_TERMINATE_PHASE
2414 self
.set_state(NetworkServiceRecordState
.VNF_TERMINATE_PHASE
)
2415 event_descr
= "Terminating VNFS in NS Id:%s" % self
.id
2416 self
.record_event("terminating-vnfs", event_descr
)
2417 yield from self
.terminate_vnfrs(self
.vnfrs
.values())
2419 # Move the state to VL_TERMINATE_PHASE
2420 self
.set_state(NetworkServiceRecordState
.VL_TERMINATE_PHASE
)
2421 event_descr
= "Terminating VLs in NS Id:%s" % self
.id
2422 self
.record_event("terminating-vls", event_descr
)
2423 yield from terminate_vlrs()
2425 yield from self
.nsm_plugin
.terminate_ns(self
)
2427 # Move the state to TERMINATED
2428 self
.set_state(NetworkServiceRecordState
.TERMINATED
)
2429 event_descr
= "Terminated NS Id:%s" % self
.id
2430 self
.record_event("terminated", event_descr
)
2433 """"Enable a NetworkServiceRecord."""
2437 """"Disable a NetworkServiceRecord."""
2440 def map_config_status(self
):
2441 self
._log
.debug("Config status for ns {} is {}".
2442 format(self
.name
, self
._config
_status
))
2443 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURING
:
2444 return 'configuring'
2445 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2449 def vl_phase_completed(self
):
2450 """ Are VLs created in this NS?"""
2451 return self
._vl
_phase
_completed
2453 def vnf_phase_completed(self
):
2454 """ Are VLs created in this NS?"""
2455 return self
._vnf
_phase
_completed
2457 def create_msg(self
):
2458 """ The network serice record as a message """
2459 nsr_dict
= {"ns_instance_config_ref": self
.id}
2460 nsr
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr
.from_dict(nsr_dict
)
2461 #nsr.cloud_account = self.cloud_account_name
2462 nsr
.sdn_account
= self
._sdn
_account
_name
2463 nsr
.name_ref
= self
.name
2464 nsr
.nsd_ref
= self
.nsd_id
2465 nsr
.nsd_name_ref
= self
.nsd_msg
.name
2466 nsr
.operational_events
= self
._op
_status
.msg
2467 nsr
.operational_status
= self
._op
_status
.yang_str()
2468 nsr
.config_status
= self
.map_config_status()
2469 nsr
.config_status_details
= self
._config
_status
_details
2470 nsr
.create_time
= self
._create
_time
2472 for cfg_prim
in self
.nsd_msg
.service_primitive
:
2473 cfg_prim
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_ServicePrimitive
.from_dict(
2475 nsr
.service_primitive
.append(cfg_prim
)
2477 for init_cfg
in self
.nsd_msg
.initial_config_primitive
:
2478 prim
= NsrYang
.NsrInitialConfigPrimitive
.from_dict(
2480 nsr
.initial_config_primitive
.append(prim
)
2482 if self
.vl_phase_completed():
2483 for vlr
in self
.vlrs
:
2484 nsr
.vlr
.append(vlr
.create_nsr_vlr_msg(self
.vnfrs
.values()))
2486 if self
.vnf_phase_completed():
2487 for vnfr_id
in self
.vnfrs
:
2488 nsr
.constituent_vnfr_ref
.append(self
.vnfrs
[vnfr_id
].const_vnfr_msg
)
2489 for vnffgr
in self
.vnffgrs
.values():
2490 nsr
.vnffgr
.append(vnffgr
.fetch_vnffgr())
2491 for scaling_group
in self
._scaling
_groups
.values():
2492 nsr
.scaling_group_record
.append(scaling_group
.create_record_msg())
2496 def all_vnfs_active(self
):
2497 """ Are all VNFS in this NS active? """
2498 for _
, vnfr
in self
.vnfrs
.items():
2499 if vnfr
.active
is not True:
2504 def update_state(self
):
2505 """ Re-evaluate this NS's state """
2506 curr_state
= self
._op
_status
.state
2508 if curr_state
== NetworkServiceRecordState
.TERMINATED
:
2509 self
._log
.debug("NS (%s) in terminated state, not updating state", self
.id)
2512 new_state
= NetworkServiceRecordState
.RUNNING
2513 self
._log
.info("Received update_state for nsr: %s, curr-state: %s",
2514 self
.id, curr_state
)
2516 # Check all the VNFRs are present
2517 for _
, vnfr
in self
.vnfrs
.items():
2518 if vnfr
.state
in [VnfRecordState
.ACTIVE
, VnfRecordState
.TERMINATED
]:
2520 elif vnfr
.state
== VnfRecordState
.FAILED
:
2521 if vnfr
._prev
_state
!= vnfr
.state
:
2522 event_descr
= "Instantiation of VNF %s failed" % vnfr
.id
2523 event_error_details
= vnfr
.state_failed_reason
2524 self
.record_event("vnf-failed", event_descr
, evt_details
=event_error_details
)
2525 vnfr
.set_state(VnfRecordState
.FAILED
)
2527 self
._log
.info("VNF state did not change, curr=%s, prev=%s",
2528 vnfr
.state
, vnfr
._prev
_state
)
2529 new_state
= NetworkServiceRecordState
.FAILED
2532 self
._log
.info("VNF %s in NSR %s is still not active; current state is: %s",
2533 vnfr
.id, self
.id, vnfr
.state
)
2534 new_state
= curr_state
2536 # If new state is RUNNING; check all VLs
2537 if new_state
== NetworkServiceRecordState
.RUNNING
:
2538 for vl
in self
.vlrs
:
2540 if vl
.state
in [VlRecordState
.ACTIVE
, VlRecordState
.TERMINATED
]:
2542 elif vl
.state
== VlRecordState
.FAILED
:
2543 if vl
.prev_state
!= vl
.state
:
2544 event_descr
= "Instantiation of VL %s failed" % vl
.id
2545 event_error_details
= vl
.state_failed_reason
2546 self
.record_event("vl-failed", event_descr
, evt_details
=event_error_details
)
2547 vl
.prev_state
= vl
.state
2549 self
._log
.debug("VL %s already in failed state")
2551 if vl
.state
in [VlRecordState
.INSTANTIATION_PENDING
, VlRecordState
.INIT
]:
2552 new_state
= NetworkServiceRecordState
.VL_INSTANTIATE
2555 if vl
.state
in [VlRecordState
.TERMINATE_PENDING
]:
2556 new_state
= NetworkServiceRecordState
.VL_TERMINATE
2559 # If new state is RUNNING; check VNFFGRs are also active
2560 if new_state
== NetworkServiceRecordState
.RUNNING
:
2561 for _
, vnffgr
in self
.vnffgrs
.items():
2562 self
._log
.info("Checking vnffgr state for nsr %s is: %s",
2563 self
.id, vnffgr
.state
)
2564 if vnffgr
.state
== VnffgRecordState
.ACTIVE
:
2566 elif vnffgr
.state
== VnffgRecordState
.FAILED
:
2567 event_descr
= "Instantiation of VNFFGR %s failed" % vnffgr
.id
2568 self
.record_event("vnffg-failed", event_descr
)
2569 new_state
= NetworkServiceRecordState
.FAILED
2572 self
._log
.info("VNFFGR %s in NSR %s is still not active; current state is: %s",
2573 vnffgr
.id, self
.id, vnffgr
.state
)
2574 new_state
= curr_state
2576 # Update all the scaling group instance operational status to
2577 # reflect the state of all VNFR within that instance
2578 yield from self
._update
_scale
_group
_instances
_status
()
2580 for _
, group
in self
._scaling
_groups
.items():
2581 if group
.state
== scale_group
.ScaleGroupState
.SCALING_OUT
:
2582 new_state
= NetworkServiceRecordState
.SCALING_OUT
2584 elif group
.state
== scale_group
.ScaleGroupState
.SCALING_IN
:
2585 new_state
= NetworkServiceRecordState
.SCALING_IN
2588 if new_state
!= curr_state
:
2589 self
._log
.debug("Changing state of Network service %s from %s to %s",
2590 self
.id, curr_state
, new_state
)
2591 if new_state
== NetworkServiceRecordState
.RUNNING
:
2592 yield from self
.is_active()
2593 elif new_state
== NetworkServiceRecordState
.FAILED
:
2594 # If the NS is already active and we entered scaling_in, scaling_out,
2595 # do not mark the NS as failing if scaling operation failed.
2596 if curr_state
in [NetworkServiceRecordState
.SCALING_OUT
,
2597 NetworkServiceRecordState
.SCALING_IN
] and self
._is
_active
:
2598 new_state
= NetworkServiceRecordState
.RUNNING
2599 self
.set_state(new_state
)
2601 yield from self
.instantiation_failed()
2603 self
.set_state(new_state
)
2605 yield from self
.publish()
2608 class InputParameterSubstitution(object):
2610 This class is responsible for substituting input parameters into an NSD.
2613 def __init__(self
, log
):
2614 """Create an instance of InputParameterSubstitution
2617 log - a logger for this object to use
2622 def __call__(self
, nsd
, nsr_config
):
2623 """Substitutes input parameters from the NSR config into the NSD
2625 This call modifies the provided NSD with the input parameters that are
2626 contained in the NSR config.
2629 nsd - a GI NSD object
2630 nsr_config - a GI NSR config object
2633 if nsd
is None or nsr_config
is None:
2636 # Create a lookup of the xpath elements that this descriptor allows
2638 optional_input_parameters
= set()
2639 for input_parameter
in nsd
.input_parameter_xpath
:
2640 optional_input_parameters
.add(input_parameter
.xpath
)
2642 # Apply the input parameters to the descriptor
2643 if nsr_config
.input_parameter
:
2644 for param
in nsr_config
.input_parameter
:
2645 if param
.xpath
not in optional_input_parameters
:
2646 msg
= "tried to set an invalid input parameter ({})"
2647 self
.log
.error(msg
.format(param
.xpath
))
2651 "input-parameter:{} = {}".format(
2658 xpath
.setxattr(nsd
, param
.xpath
, param
.value
)
2660 except Exception as e
:
2661 self
.log
.exception(e
)
2664 class NetworkServiceDescriptor(object):
2666 Network service descriptor class
2669 def __init__(self
, dts
, log
, loop
, nsd
, nsm
):
2681 """ Returns nsd id """
2686 """ Returns name of nsd """
2687 return self
._nsd
.name
2690 def ref_count(self
):
2691 """ Returns reference count"""
2692 return self
._ref
_count
2695 """ Returns whether nsd is in use or not """
2696 return True if self
.ref_count
> 0 else False
2699 """ Take a reference on this object """
2700 self
._ref
_count
+= 1
2703 """ Release reference on this object """
2704 if self
.ref_count
< 1:
2705 msg
= ("Unref on a NSD object - nsd id %s, ref_count = %s" %
2706 (self
.id, self
.ref_count
))
2707 self
._log
.critical(msg
)
2708 raise NetworkServiceDescriptorError(msg
)
2709 self
._ref
_count
-= 1
2713 """ Return the message associated with this NetworkServiceDescriptor"""
2717 def path_for_id(nsd_id
):
2718 """ Return path for the passed nsd_id"""
2719 return "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}'".format(nsd_id
)
2722 """ Return the message associated with this NetworkServiceDescriptor"""
2723 return NetworkServiceDescriptor
.path_for_id(self
.id)
2725 def update(self
, nsd
):
2726 """ Update the NSD descriptor """
2730 class NsdDtsHandler(object):
2731 """ The network service descriptor DTS handler """
2732 XPATH
= "C,/nsd:nsd-catalog/nsd:nsd"
2734 def __init__(self
, dts
, log
, loop
, nsm
):
2744 """ Return registration handle """
2749 """ Register for Nsd create/update/delete/read requests from dts """
2751 def on_apply(dts
, acg
, xact
, action
, scratch
):
2752 """Apply the configuration"""
2753 is_recovery
= xact
.xact
is None and action
== rwdts
.AppconfAction
.INSTALL
2754 self
._log
.debug("Got nsd apply cfg (xact:%s) (action:%s)",
2756 # Create/Update an NSD record
2757 for cfg
in self
._regh
.get_xact_elements(xact
):
2758 # Only interested in those NSD cfgs whose ID was received in prepare callback
2759 if cfg
.id in scratch
.get('nsds', []) or is_recovery
:
2760 self
._nsm
.update_nsd(cfg
)
2762 scratch
.pop('nsds', None)
2764 return RwTypes
.RwStatus
.SUCCESS
2767 def delete_nsd_libs(nsd_id
):
2768 """ Remove any files uploaded with NSD and stored under $RIFT_ARTIFACTS/libs/<id> """
2770 rift_artifacts_dir
= os
.environ
['RIFT_ARTIFACTS']
2771 nsd_dir
= os
.path
.join(rift_artifacts_dir
, 'launchpad/libs', nsd_id
)
2773 if os
.path
.exists (nsd_dir
):
2774 shutil
.rmtree(nsd_dir
, ignore_errors
=True)
2775 except Exception as e
:
2776 self
._log
.error("Exception in cleaning up NSD libs {}: {}".
2778 self
._log
.excpetion(e
)
2781 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2782 """ Prepare callback from DTS for NSD config """
2784 self
._log
.info("Got nsd prepare - config received nsd id %s, msg %s",
2787 fref
= ProtobufC
.FieldReference
.alloc()
2788 fref
.goto_whole_message(msg
.to_pbcm())
2790 if fref
.is_field_deleted():
2791 # Delete an NSD record
2792 self
._log
.debug("Deleting NSD with id %s", msg
.id)
2793 if self
._nsm
.nsd_in_use(msg
.id):
2794 self
._log
.debug("Cannot delete NSD in use - %s", msg
.id)
2795 err
= "Cannot delete an NSD in use - %s" % msg
.id
2796 raise NetworkServiceDescriptorRefCountExists(err
)
2798 yield from delete_nsd_libs(msg
.id)
2799 self
._nsm
.delete_nsd(msg
.id)
2801 # Add this NSD to scratch to create/update in apply callback
2802 nsds
= scratch
.setdefault('nsds', [])
2804 # acg._scratch['nsds'].append(msg.id)
2806 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2809 "Registering for NSD config using xpath: %s",
2810 NsdDtsHandler
.XPATH
,
2813 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2814 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2815 # Need a list in scratch to store NSDs to create/update later
2816 # acg._scratch['nsds'] = list()
2817 self
._regh
= acg
.register(
2818 xpath
=NsdDtsHandler
.XPATH
,
2819 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
2820 on_prepare
=on_prepare
)
2823 class VnfdDtsHandler(object):
2824 """ DTS handler for VNFD config changes """
2825 XPATH
= "C,/vnfd:vnfd-catalog/vnfd:vnfd"
2827 def __init__(self
, dts
, log
, loop
, nsm
):
2836 """ DTS registration handle """
2841 """ Register for VNFD configuration"""
2844 def on_apply(dts
, acg
, xact
, action
, scratch
):
2845 """Apply the configuration"""
2846 self
._log
.debug("Got NSM VNFD apply (xact: %s) (action: %s)(scr: %s)",
2847 xact
, action
, scratch
)
2849 # Create/Update a VNFD record
2850 for cfg
in self
._regh
.get_xact_elements(xact
):
2851 # Only interested in those VNFD cfgs whose ID was received in prepare callback
2852 if cfg
.id in scratch
.get('vnfds', []):
2853 self
._nsm
.update_vnfd(cfg
)
2855 for cfg
in self
._regh
.elements
:
2856 if cfg
.id in scratch
.get('deleted_vnfds', []):
2857 yield from self
._nsm
.delete_vnfd(cfg
.id)
2859 scratch
.pop('vnfds', None)
2860 scratch
.pop('deleted_vnfds', None)
2863 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2864 """ on prepare callback """
2865 self
._log
.debug("Got on prepare for VNFD (path: %s) (action: %s) (msg: %s)",
2866 ks_path
.to_xpath(RwNsmYang
.get_schema()), xact_info
.query_action
, msg
)
2868 fref
= ProtobufC
.FieldReference
.alloc()
2869 fref
.goto_whole_message(msg
.to_pbcm())
2871 # Handle deletes in prepare_callback, but adds/updates in apply_callback
2872 if fref
.is_field_deleted():
2873 self
._log
.debug("Adding msg to deleted field")
2874 deleted_vnfds
= scratch
.setdefault('deleted_vnfds', [])
2875 deleted_vnfds
.append(msg
.id)
2877 # Add this VNFD to scratch to create/update in apply callback
2878 vnfds
= scratch
.setdefault('vnfds', [])
2879 vnfds
.append(msg
.id)
2881 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2884 "Registering for VNFD config using xpath: %s",
2885 VnfdDtsHandler
.XPATH
,
2887 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2888 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2889 # Need a list in scratch to store VNFDs to create/update later
2890 # acg._scratch['vnfds'] = list()
2891 # acg._scratch['deleted_vnfds'] = list()
2892 self
._regh
= acg
.register(
2893 xpath
=VnfdDtsHandler
.XPATH
,
2894 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY
,
2895 on_prepare
=on_prepare
)
2897 class NsrRpcDtsHandler(object):
2898 """ The network service instantiation RPC DTS handler """
2899 EXEC_NSR_CONF_XPATH
= "I,/nsr:start-network-service"
2900 EXEC_NSR_CONF_O_XPATH
= "O,/nsr:start-network-service"
2901 NETCONF_IP_ADDRESS
= "127.0.0.1"
2903 RESTCONF_PORT
= 8888
2904 NETCONF_USER
= "admin"
2905 NETCONF_PW
= "admin"
2906 REST_BASE_V2_URL
= 'https://{}:{}/v2/api/'.format("127.0.0.1",8888)
2908 def __init__(self
, dts
, log
, loop
, nsm
):
2915 self
._ns
_regh
= None
2917 self
._manager
= None
2918 self
._nsr
_config
_url
= NsrRpcDtsHandler
.REST_BASE_V2_URL
+ 'config/ns-instance-config'
2920 self
._model
= RwYang
.Model
.create_libncx()
2921 self
._model
.load_schema_ypbc(RwNsrYang
.get_schema())
2925 """ Return the NS manager instance """
2929 def wrap_netconf_config_xml(xml
):
2930 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
2934 def _connect(self
, timeout_secs
=240):
2936 start_time
= time
.time()
2937 while (time
.time() - start_time
) < timeout_secs
:
2940 self
._log
.debug("Attemping NsmTasklet netconf connection.")
2942 manager
= yield from ncclient
.asyncio_manager
.asyncio_connect(
2944 host
=NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
,
2945 port
=NsrRpcDtsHandler
.NETCONF_PORT
,
2946 username
=NsrRpcDtsHandler
.NETCONF_USER
,
2947 password
=NsrRpcDtsHandler
.NETCONF_PW
,
2949 look_for_keys
=False,
2950 hostkey_verify
=False,
2955 except ncclient
.transport
.errors
.SSHError
as e
:
2956 self
._log
.warning("Netconf connection to launchpad %s failed: %s",
2957 NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
, str(e
))
2959 yield from asyncio
.sleep(5, loop
=self
._loop
)
2961 raise NsrInstantiationFailed("Failed to connect to Launchpad within %s seconds" %
2964 def _apply_ns_instance_config(self
,payload_dict
):
2965 #self._log.debug("At apply NS instance config with payload %s",payload_dict)
2966 req_hdr
= {'accept':'application/vnd.yang.data+json','content-type':'application/vnd.yang.data+json'}
2967 response
=requests
.post(self
._nsr
_config
_url
, headers
=req_hdr
, auth
=('admin', 'admin'),data
=payload_dict
,verify
=False)
2972 """ Register for NS monitoring read from dts """
2974 def on_ns_config_prepare(xact_info
, action
, ks_path
, msg
):
2975 """ prepare callback from dts start-network-service"""
2976 assert action
== rwdts
.QueryAction
.RPC
2978 rpc_op
= NsrYang
.YangOutput_Nsr_StartNetworkService
.from_dict({
2979 "nsr_id":str(uuid
.uuid4())
2982 if not ('name' in rpc_ip
and 'nsd_ref' in rpc_ip
and ('cloud_account' in rpc_ip
or 'om_datacenter' in rpc_ip
)):
2983 self
._log
.error("Mandatory parameters name or nsd_ref or cloud account not found in start-network-service {}".format(rpc_ip
))
2986 self
._log
.debug("start-network-service RPC input: {}".format(rpc_ip
))
2989 # Add used value to the pool
2990 self
._log
.debug("RPC output: {}".format(rpc_op
))
2992 nsd_copy
= self
.nsm
.get_nsd(rpc_ip
.nsd_ref
)
2994 #if not self._manager:
2995 # self._manager = yield from self._connect()
2997 self
._log
.debug("Configuring ns-instance-config with name %s nsd-ref: %s",
2998 rpc_ip
.name
, rpc_ip
.nsd_ref
)
3000 ns_instance_config_dict
= {"id":rpc_op
.nsr_id
, "admin_status":"ENABLED"}
3001 ns_instance_config_copy_dict
= {k
:v
for k
, v
in rpc_ip
.as_dict().items()
3002 if k
in RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr().fields
}
3003 ns_instance_config_dict
.update(ns_instance_config_copy_dict
)
3005 ns_instance_config
= RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.from_dict(ns_instance_config_dict
)
3006 ns_instance_config
.nsd
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_Nsd()
3007 ns_instance_config
.nsd
.from_dict(nsd_copy
.msg
.as_dict())
3009 payload_dict
= ns_instance_config
.to_json(self
._model
)
3010 #xml = ns_instance_config.to_xml_v2(self._model)
3011 #netconf_xml = self.wrap_netconf_config_xml(xml)
3013 #self._log.debug("Sending configure ns-instance-config xml to %s: %s",
3014 # netconf_xml, NsrRpcDtsHandler.NETCONF_IP_ADDRESS)
3015 self
._log
.debug("Sending configure ns-instance-config json to %s: %s",
3016 self
._nsr
_config
_url
,ns_instance_config
)
3018 #response = yield from self._manager.edit_config(
3020 # config=netconf_xml,
3022 response
= yield from self
._loop
.run_in_executor(
3024 self
._apply
_ns
_instance
_config
,
3027 response
.raise_for_status()
3028 self
._log
.debug("Received edit config response: %s", response
.json())
3030 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
,
3031 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
,
3033 except Exception as e
:
3034 self
._log
.error("Exception processing the "
3035 "start-network-service: {}".format(e
))
3036 self
._log
.exception(e
)
3037 xact_info
.respond_xpath(rwdts
.XactRspCode
.NACK
,
3038 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
)
3041 hdl_ns
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_ns_config_prepare
,)
3043 with self
._dts
.group_create() as group
:
3044 self
._ns
_regh
= group
.register(xpath
=NsrRpcDtsHandler
.EXEC_NSR_CONF_XPATH
,
3046 flags
=rwdts
.Flag
.PUBLISHER
,
3050 class NsrDtsHandler(object):
3051 """ The network service DTS handler """
3052 NSR_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr"
3053 SCALE_INSTANCE_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr/nsr:scaling-group/nsr:instance"
3054 KEY_PAIR_XPATH
= "C,/nsr:key-pair"
3056 def __init__(self
, dts
, log
, loop
, nsm
):
3062 self
._nsr
_regh
= None
3063 self
._scale
_regh
= None
3064 self
._key
_pair
_regh
= None
3068 """ Return the NS manager instance """
3073 """ Register for Nsr create/update/delete/read requests from dts """
3075 def nsr_id_from_keyspec(ks
):
3076 nsr_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.schema().keyspec_to_entry(ks
)
3077 nsr_id
= nsr_path_entry
.key00
.id
3080 def group_name_from_keyspec(ks
):
3081 group_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
.schema().keyspec_to_entry(ks
)
3082 group_name
= group_path_entry
.key00
.scaling_group_name_ref
3085 def is_instance_in_reg_elements(nsr_id
, group_name
, instance_id
):
3086 """ Return boolean indicating if scaling group instance was already commited previously.
3088 By looking at the existing elements in this registration handle (elements not part
3089 of this current xact), we can tell if the instance was configured previously without
3090 keeping any application state.
3092 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3093 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3094 elem_group_name
= group_name_from_keyspec(keyspec
)
3096 if elem_nsr_id
!= nsr_id
or group_name
!= elem_group_name
:
3099 if instance_cfg
.id == instance_id
:
3104 def get_scale_group_instance_delta(nsr_id
, group_name
, xact
):
3105 delta
= {"added": [], "deleted": []}
3106 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3107 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3108 if elem_nsr_id
!= nsr_id
:
3111 elem_group_name
= group_name_from_keyspec(keyspec
)
3112 if elem_group_name
!= group_name
:
3115 delta
["added"].append(instance_cfg
.id)
3117 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(include_keyspec
=True):
3118 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3119 if elem_nsr_id
!= nsr_id
:
3122 elem_group_name
= group_name_from_keyspec(keyspec
)
3123 if elem_group_name
!= group_name
:
3126 if instance_cfg
.id in delta
["added"]:
3127 delta
["added"].remove(instance_cfg
.id)
3129 delta
["deleted"].append(instance_cfg
.id)
3134 def update_nsr_nsd(nsr_id
, xact
, scratch
):
3137 def get_nsr_vl_delta(nsr_id
, xact
, scratch
):
3138 delta
= {"added": [], "deleted": []}
3139 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3140 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3141 if elem_nsr_id
!= nsr_id
:
3144 if 'vld' in instance_cfg
.nsd
:
3145 for vld
in instance_cfg
.nsd
.vld
:
3146 delta
["added"].append(vld
)
3148 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3149 self
._log
.debug("NSR update: %s", instance_cfg
)
3150 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3151 if elem_nsr_id
!= nsr_id
:
3154 if 'vld' in instance_cfg
.nsd
:
3155 for vld
in instance_cfg
.nsd
.vld
:
3156 if vld
in delta
["added"]:
3157 delta
["added"].remove(vld
)
3159 delta
["deleted"].append(vld
)
3163 vl_delta
= yield from get_nsr_vl_delta(nsr_id
, xact
, scratch
)
3164 self
._log
.debug("Got NSR:%s VL instance delta: %s", nsr_id
, vl_delta
)
3166 for vld
in vl_delta
["added"]:
3167 yield from self
._nsm
.nsr_instantiate_vl(nsr_id
, vld
)
3169 for vld
in vl_delta
["deleted"]:
3170 yield from self
._nsm
.nsr_terminate_vl(nsr_id
, vld
)
3172 def get_add_delete_update_cfgs(dts_member_reg
, xact
, key_name
, scratch
):
3173 # Unfortunately, it is currently difficult to figure out what has exactly
3174 # changed in this xact without Pbdelta support (RIFT-4916)
3175 # As a workaround, we can fetch the pre and post xact elements and
3176 # perform a comparison to figure out adds/deletes/updates
3177 xact_cfgs
= list(dts_member_reg
.get_xact_elements(xact
))
3178 curr_cfgs
= list(dts_member_reg
.elements
)
3180 xact_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in xact_cfgs
}
3181 curr_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in curr_cfgs
}
3184 added_keys
= set(xact_key_map
) - set(curr_key_map
)
3185 added_cfgs
= [xact_key_map
[key
] for key
in added_keys
]
3188 deleted_keys
= set(curr_key_map
) - set(xact_key_map
)
3189 deleted_cfgs
= [curr_key_map
[key
] for key
in deleted_keys
]
3192 updated_keys
= set(curr_key_map
) & set(xact_key_map
)
3193 updated_cfgs
= [xact_key_map
[key
] for key
in updated_keys
3194 if xact_key_map
[key
] != curr_key_map
[key
]]
3196 return added_cfgs
, deleted_cfgs
, updated_cfgs
3198 def get_nsr_key_pairs(dts_member_reg
, xact
):
3200 for instance_cfg
, keyspec
in dts_member_reg
.get_xact_elements(xact
, include_keyspec
=True):
3201 self
._log
.debug("Key pair received is {} KS: {}".format(instance_cfg
, keyspec
))
3202 xpath
= keyspec
.to_xpath(RwNsrYang
.get_schema())
3203 key_pairs
[instance_cfg
.name
] = instance_cfg
3206 def on_apply(dts
, acg
, xact
, action
, scratch
):
3207 """Apply the configuration"""
3208 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3209 xact
, action
, scratch
)
3211 def handle_create_nsr(msg
, key_pairs
=None, restart_mode
=False):
3212 # Handle create nsr requests """
3213 # Do some validations
3214 if not msg
.has_field("nsd"):
3215 err
= "NSD not provided"
3216 self
._log
.error(err
)
3217 raise NetworkServiceRecordError(err
)
3219 self
._log
.debug("Creating NetworkServiceRecord %s from nsr config %s",
3220 msg
.id, msg
.as_dict())
3221 nsr
= self
.nsm
.create_nsr(msg
, key_pairs
=key_pairs
, restart_mode
=restart_mode
)
3224 def handle_delete_nsr(msg
):
3226 def delete_instantiation(ns_id
):
3227 """ Delete instantiation """
3228 with self
._dts
.transaction() as xact
:
3229 yield from self
._nsm
.terminate_ns(ns_id
, xact
)
3231 # Handle delete NSR requests
3232 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3233 # Terminate the NSR instance
3234 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3236 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3237 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3238 nsr
.record_event("terminate-rcvd", event_descr
)
3240 self
._loop
.create_task(delete_instantiation(msg
.id))
3243 def begin_instantiation(nsr
):
3244 # Begin instantiation
3245 self
._log
.info("Beginning NS instantiation: %s", nsr
.id)
3246 yield from self
._nsm
.instantiate_ns(nsr
.id, xact
)
3248 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3249 xact
, action
, scratch
)
3251 if action
== rwdts
.AppconfAction
.INSTALL
and xact
.id is None:
3253 for element
in self
._key
_pair
_regh
.elements
:
3254 key_pairs
.append(element
)
3255 for element
in self
._nsr
_regh
.elements
:
3256 nsr
= handle_create_nsr(element
, key_pairs
, restart_mode
=True)
3257 self
._loop
.create_task(begin_instantiation(nsr
))
3260 (added_msgs
, deleted_msgs
, updated_msgs
) = get_add_delete_update_cfgs(self
._nsr
_regh
,
3264 self
._log
.debug("Added: %s, Deleted: %s, Updated: %s", added_msgs
,
3265 deleted_msgs
, updated_msgs
)
3267 for msg
in added_msgs
:
3268 if msg
.id not in self
._nsm
.nsrs
:
3269 self
._log
.info("Create NSR received in on_apply to instantiate NS:%s", msg
.id)
3270 key_pairs
= get_nsr_key_pairs(self
._key
_pair
_regh
, xact
)
3271 nsr
= handle_create_nsr(msg
,key_pairs
)
3272 self
._loop
.create_task(begin_instantiation(nsr
))
3274 for msg
in deleted_msgs
:
3275 self
._log
.info("Delete NSR received in on_apply to terminate NS:%s", msg
.id)
3277 handle_delete_nsr(msg
)
3279 self
._log
.exception("Failed to terminate NS:%s", msg
.id)
3281 for msg
in updated_msgs
:
3282 self
._log
.info("Update NSR received in on_apply: %s", msg
)
3284 self
._nsm
.nsr_update_cfg(msg
.id, msg
)
3287 self
._loop
.create_task(update_nsr_nsd(msg
.id, xact
, scratch
))
3289 for group
in msg
.scaling_group
:
3290 instance_delta
= get_scale_group_instance_delta(msg
.id, group
.scaling_group_name_ref
, xact
)
3291 self
._log
.debug("Got NSR:%s scale group instance delta: %s", msg
.id, instance_delta
)
3293 for instance_id
in instance_delta
["added"]:
3294 self
._nsm
.scale_nsr_out(msg
.id, group
.scaling_group_name_ref
, instance_id
, xact
)
3296 for instance_id
in instance_delta
["deleted"]:
3297 self
._nsm
.scale_nsr_in(msg
.id, group
.scaling_group_name_ref
, instance_id
)
3300 return RwTypes
.RwStatus
.SUCCESS
3303 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
3304 """ Prepare calllback from DTS for NSR """
3306 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3307 action
= xact_info
.query_action
3309 "Got Nsr prepare callback (xact: %s) (action: %s) (info: %s), %s:%s)",
3310 xact
, action
, xact_info
, xpath
, msg
3314 def delete_instantiation(ns_id
):
3315 """ Delete instantiation """
3316 yield from self
._nsm
.terminate_ns(ns_id
, None)
3318 def handle_delete_nsr():
3319 """ Handle delete NSR requests """
3320 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3321 # Terminate the NSR instance
3322 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3324 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3325 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3326 nsr
.record_event("terminate-rcvd", event_descr
)
3328 self
._loop
.create_task(delete_instantiation(msg
.id))
3330 fref
= ProtobufC
.FieldReference
.alloc()
3331 fref
.goto_whole_message(msg
.to_pbcm())
3333 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
, rwdts
.QueryAction
.DELETE
]:
3334 # if this is an NSR create
3335 if action
!= rwdts
.QueryAction
.DELETE
and msg
.id not in self
._nsm
.nsrs
:
3336 # Ensure the Cloud account/datacenter has been specified
3337 if not msg
.has_field("cloud_account") and not msg
.has_field("om_datacenter"):
3338 raise NsrInstantiationFailed("Cloud account or datacenter not specified in NSR")
3340 # Check if nsd is specified
3341 if not msg
.has_field("nsd"):
3342 raise NsrInstantiationFailed("NSD not specified in NSR")
3345 nsr
= self
._nsm
.nsrs
[msg
.id]
3347 if msg
.has_field("nsd"):
3348 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3349 raise NsrVlUpdateError("Unable to update VL when NSR not in running state")
3350 if 'vld' not in msg
.nsd
or len(msg
.nsd
.vld
) == 0:
3351 raise NsrVlUpdateError("NS config NSD should have atleast 1 VLD defined")
3353 if msg
.has_field("scaling_group"):
3354 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3355 raise ScalingOperationError("Unable to perform scaling action when NS is not in running state")
3357 if len(msg
.scaling_group
) > 1:
3358 raise ScalingOperationError("Only a single scaling group can be configured at a time")
3360 for group_msg
in msg
.scaling_group
:
3361 num_new_group_instances
= len(group_msg
.instance
)
3362 if num_new_group_instances
> 1:
3363 raise ScalingOperationError("Only a single scaling instance can be modified at a time")
3365 elif num_new_group_instances
== 1:
3366 scale_group
= nsr
.scaling_groups
[group_msg
.scaling_group_name_ref
]
3367 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
]:
3368 if len(scale_group
.instances
) == scale_group
.max_instance_count
:
3369 raise ScalingOperationError("Max instances for %s reached" % scale_group
)
3371 acg
.handle
.prepare_complete_ok(xact_info
.handle
)
3374 self
._log
.debug("Registering for NSR config using xpath: %s",
3375 NsrDtsHandler
.NSR_XPATH
)
3377 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
3378 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
3379 self
._nsr
_regh
= acg
.register(xpath
=NsrDtsHandler
.NSR_XPATH
,
3380 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3381 on_prepare
=on_prepare
)
3383 self
._scale
_regh
= acg
.register(
3384 xpath
=NsrDtsHandler
.SCALE_INSTANCE_XPATH
,
3385 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY| rwdts
.Flag
.CACHE
,
3388 self
._key
_pair
_regh
= acg
.register(
3389 xpath
=NsrDtsHandler
.KEY_PAIR_XPATH
,
3390 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3394 class NsrOpDataDtsHandler(object):
3395 """ The network service op data DTS handler """
3396 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
3398 def __init__(self
, dts
, log
, loop
, nsm
):
3407 """ Return the registration handle"""
3412 """ Return the NS manager instance """
3417 """ Register for Nsr op data publisher registration"""
3418 self
._log
.debug("Registering Nsr op data path %s as publisher",
3419 NsrOpDataDtsHandler
.XPATH
)
3421 hdl
= rift
.tasklets
.DTS
.RegistrationHandler()
3422 handlers
= rift
.tasklets
.Group
.Handler()
3423 with self
._dts
.group_create(handler
=handlers
) as group
:
3424 self
._regh
= group
.register(xpath
=NsrOpDataDtsHandler
.XPATH
,
3426 flags
=rwdts
.Flag
.PUBLISHER | rwdts
.Flag
.NO_PREP_READ | rwdts
.Flag
.DATASTORE
)
3429 def create(self
, path
, msg
):
3431 Create an NS record in DTS with the path and message
3433 self
._log
.debug("Creating NSR %s:%s", path
, msg
)
3434 self
.regh
.create_element(path
, msg
)
3435 self
._log
.debug("Created NSR, %s:%s", path
, msg
)
3438 def update(self
, path
, msg
, flags
=rwdts
.XactFlag
.REPLACE
):
3440 Update an NS record in DTS with the path and message
3442 self
._log
.debug("Updating NSR, %s:%s regh = %s", path
, msg
, self
.regh
)
3443 self
.regh
.update_element(path
, msg
, flags
)
3444 self
._log
.debug("Updated NSR, %s:%s", path
, msg
)
3447 def delete(self
, path
):
3449 Update an NS record in DTS with the path and message
3451 self
._log
.debug("Deleting NSR path:%s", path
)
3452 self
.regh
.delete_element(path
)
3453 self
._log
.debug("Deleted NSR path:%s", path
)
3456 class VnfrDtsHandler(object):
3457 """ The virtual network service DTS handler """
3458 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
3460 def __init__(self
, dts
, log
, loop
, nsm
):
3470 """ Return registration handle """
3475 """ Return the NS manager instance """
3480 """ Register for vnfr create/update/delete/ advises from dts """
3482 def on_commit(xact_info
):
3483 """ The transaction has been committed """
3484 self
._log
.debug("Got vnfr commit (xact_info: %s)", xact_info
)
3485 return rwdts
.MemberRspCode
.ACTION_OK
3488 def on_prepare(xact_info
, action
, ks_path
, msg
):
3489 """ prepare callback from dts """
3490 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3492 "Got vnfr on_prepare cb (xact_info: %s, action: %s): %s:%s",
3493 xact_info
, action
, ks_path
, msg
3496 schema
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr
.schema()
3497 path_entry
= schema
.keyspec_to_entry(ks_path
)
3498 if path_entry
.key00
.id not in self
._nsm
._vnfrs
:
3499 self
._log
.error("%s request for non existent record path %s",
3501 xact_info
.respond_xpath(rwdts
.XactRspCode
.NA
, xpath
)
3505 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3506 if action
== rwdts
.QueryAction
.CREATE
or action
== rwdts
.QueryAction
.UPDATE
:
3507 yield from self
._nsm
.update_vnfr(msg
)
3508 elif action
== rwdts
.QueryAction
.DELETE
:
3509 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3510 self
._nsm
.delete_vnfr(path_entry
.key00
.id)
3512 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
, xpath
)
3514 self
._log
.debug("Registering for VNFR using xpath: %s",
3515 VnfrDtsHandler
.XPATH
,)
3517 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_commit
=on_commit
,
3518 on_prepare
=on_prepare
,)
3519 with self
._dts
.group_create() as group
:
3520 self
._regh
= group
.register(xpath
=VnfrDtsHandler
.XPATH
,
3522 flags
=(rwdts
.Flag
.SUBSCRIBER
),)
3525 class NsdRefCountDtsHandler(object):
3526 """ The NSD Ref Count DTS handler """
3527 XPATH
= "D,/nsr:ns-instance-opdata/rw-nsr:nsd-ref-count"
3529 def __init__(self
, dts
, log
, loop
, nsm
):
3539 """ Return registration handle """
3544 """ Return the NS manager instance """
3549 """ Register for NSD ref count read from dts """
3552 def on_prepare(xact_info
, action
, ks_path
, msg
):
3553 """ prepare callback from dts """
3554 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3556 if action
== rwdts
.QueryAction
.READ
:
3557 schema
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount
.schema()
3558 path_entry
= schema
.keyspec_to_entry(ks_path
)
3559 nsd_list
= yield from self
._nsm
.get_nsd_refcount(path_entry
.key00
.nsd_id_ref
)
3560 for xpath
, msg
in nsd_list
:
3561 xact_info
.respond_xpath(rsp_code
=rwdts
.XactRspCode
.MORE
,
3564 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
3566 raise NetworkServiceRecordError("Not supported operation %s" % action
)
3568 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_prepare
,)
3569 with self
._dts
.group_create() as group
:
3570 self
._regh
= group
.register(xpath
=NsdRefCountDtsHandler
.XPATH
,
3572 flags
=rwdts
.Flag
.PUBLISHER
,)
3575 class NsManager(object):
3576 """ The Network Service Manager class"""
3577 def __init__(self
, dts
, log
, loop
,
3578 nsr_handler
, vnfr_handler
, vlr_handler
, ro_plugin_selector
,
3579 vnffgmgr
, vnfd_pub_handler
, cloud_account_handler
):
3583 self
._nsr
_handler
= nsr_handler
3584 self
._vnfr
_pub
_handler
= vnfr_handler
3585 self
._vlr
_pub
_handler
= vlr_handler
3586 self
._vnffgmgr
= vnffgmgr
3587 self
._vnfd
_pub
_handler
= vnfd_pub_handler
3588 self
._cloud
_account
_handler
= cloud_account_handler
3590 self
._ro
_plugin
_selector
= ro_plugin_selector
3591 self
._ncclient
= rift
.mano
.ncclient
.NcClient(
3603 self
.cfgmgr_obj
= conman
.ROConfigManager(log
, loop
, dts
, self
)
3605 # TODO: All these handlers should move to tasklet level.
3606 # Passing self is often an indication of bad design
3607 self
._nsd
_dts
_handler
= NsdDtsHandler(dts
, log
, loop
, self
)
3608 self
._vnfd
_dts
_handler
= VnfdDtsHandler(dts
, log
, loop
, self
)
3609 self
._dts
_handlers
= [self
._nsd
_dts
_handler
,
3610 VnfrDtsHandler(dts
, log
, loop
, self
),
3611 NsdRefCountDtsHandler(dts
, log
, loop
, self
),
3612 NsrDtsHandler(dts
, log
, loop
, self
),
3613 ScalingRpcHandler(log
, dts
, loop
, self
.scale_rpc_callback
),
3614 NsrRpcDtsHandler(dts
,log
,loop
,self
),
3615 self
._vnfd
_dts
_handler
,
3636 def nsr_handler(self
):
3637 """" NSR handler """
3638 return self
._nsr
_handler
3642 """" So Obj handler """
3647 """ NSRs in this NSM"""
3652 """ NSDs in this NSM"""
3657 """ VNFDs in this NSM"""
3662 """ VNFRs in this NSM"""
3666 def nsr_pub_handler(self
):
3667 """ NSR publication handler """
3668 return self
._nsr
_handler
3671 def vnfr_pub_handler(self
):
3672 """ VNFR publication handler """
3673 return self
._vnfr
_pub
_handler
3676 def vlr_pub_handler(self
):
3677 """ VLR publication handler """
3678 return self
._vlr
_pub
_handler
3681 def vnfd_pub_handler(self
):
3682 return self
._vnfd
_pub
_handler
3686 """ Register all static DTS handlers """
3687 for dts_handle
in self
._dts
_handlers
:
3688 yield from dts_handle
.register()
3691 def get_ns_by_nsr_id(self
, nsr_id
):
3692 """ get NSR by nsr id """
3693 if nsr_id
not in self
._nsrs
:
3694 raise NetworkServiceRecordError("NSR id %s not found" % nsr_id
)
3696 return self
._nsrs
[nsr_id
]
3698 def scale_nsr_out(self
, nsr_id
, scale_group_name
, instance_id
, config_xact
):
3699 self
.log
.debug("Scale out NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3704 nsr
= self
._nsrs
[nsr_id
]
3705 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3706 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3708 self
._loop
.create_task(nsr
.create_scale_group_instance(scale_group_name
, instance_id
, config_xact
))
3710 def scale_nsr_in(self
, nsr_id
, scale_group_name
, instance_id
):
3711 self
.log
.debug("Scale in NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3716 nsr
= self
._nsrs
[nsr_id
]
3717 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3718 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3720 self
._loop
.create_task(nsr
.delete_scale_group_instance(scale_group_name
, instance_id
))
3722 def scale_rpc_callback(self
, xact
, msg
, action
):
3723 """Callback handler for RPC calls
3725 xact : Transaction Handler
3727 action : Scaling Action
3729 ScalingGroupInstance
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup_Instance
3730 ScalingGroup
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
3732 xpath
= ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').format(
3734 instance
= ScalingGroupInstance
.from_dict({"id": msg
.instance_id
})
3737 def get_nsr_scaling_group():
3738 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3740 for result
in results
:
3741 res
= yield from result
3742 nsr_config
= res
.result
3744 for scaling_group
in nsr_config
.scaling_group
:
3745 if scaling_group
.scaling_group_name_ref
== msg
.scaling_group_name_ref
:
3748 scaling_group
= nsr_config
.scaling_group
.add()
3749 scaling_group
.scaling_group_name_ref
= msg
.scaling_group_name_ref
3751 return (nsr_config
, scaling_group
)
3754 def update_config(nsr_config
):
3755 xml
= self
._ncclient
.convert_to_xml(RwNsrYang
, nsr_config
)
3756 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
3757 yield from self
._ncclient
.connect()
3758 yield from self
._ncclient
.manager
.edit_config(target
="running", config
=xml
, default_operation
="replace")
3762 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3763 scaling_group
.instance
.append(instance
)
3764 yield from update_config(nsr_config
)
3768 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3769 scaling_group
.instance
.remove(instance
)
3770 yield from update_config(nsr_config
)
3772 if action
== ScalingRpcHandler
.ACTION
.SCALE_OUT
:
3773 self
._loop
.create_task(scale_out())
3775 self
._loop
.create_task(scale_in())
3777 # Opdata based calls, disabled for now!
3778 # if action == ScalingRpcHandler.ACTION.SCALE_OUT:
3779 # self.scale_nsr_out(
3781 # msg.scaling_group_name_ref,
3785 # self.scale_nsr_in(
3787 # msg.scaling_group_name_ref,
3790 def nsr_update_cfg(self
, nsr_id
, msg
):
3791 nsr
= self
._nsrs
[nsr_id
]
3792 nsr
.nsr_cfg_msg
= msg
3794 def nsr_instantiate_vl(self
, nsr_id
, vld
):
3795 self
.log
.debug("NSR {} create VL {}".format(nsr_id
, vld
))
3796 nsr
= self
._nsrs
[nsr_id
]
3797 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3798 raise NsrVlUpdateError("Cannot perform VL instantiate if NSR is not in running state")
3800 # Not calling in a separate task as this is called from a separate task
3801 yield from nsr
.create_vl_instance(vld
)
3803 def nsr_terminate_vl(self
, nsr_id
, vld
):
3804 self
.log
.debug("NSR {} delete VL {}".format(nsr_id
, vld
.id))
3805 nsr
= self
._nsrs
[nsr_id
]
3806 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3807 raise NsrVlUpdateError("Cannot perform VL terminate if NSR is not in running state")
3809 # Not calling in a separate task as this is called from a separate task
3810 yield from nsr
.delete_vl_instance(vld
)
3812 def create_nsr(self
, nsr_msg
, key_pairs
=None,restart_mode
=False):
3813 """ Create an NSR instance """
3814 if nsr_msg
.id in self
._nsrs
:
3815 msg
= "NSR id %s already exists" % nsr_msg
.id
3816 self
._log
.error(msg
)
3817 raise NetworkServiceRecordError(msg
)
3819 self
._log
.info("Create NetworkServiceRecord nsr id %s from nsd_id %s",
3823 nsm_plugin
= self
._ro
_plugin
_selector
.ro_plugin
3824 sdn_account_name
= self
._cloud
_account
_handler
.get_cloud_account_sdn_name(nsr_msg
.cloud_account
)
3826 nsr
= NetworkServiceRecord(self
._dts
,
3834 restart_mode
=restart_mode
3836 self
._nsrs
[nsr_msg
.id] = nsr
3837 nsm_plugin
.create_nsr(nsr_msg
, nsr_msg
.nsd
, key_pairs
)
3841 def delete_nsr(self
, nsr_id
):
3843 Delete NSR with the passed nsr id
3845 del self
._nsrs
[nsr_id
]
3848 def instantiate_ns(self
, nsr_id
, config_xact
):
3849 """ Instantiate an NS instance """
3850 self
._log
.debug("Instantiating Network service id %s", nsr_id
)
3851 if nsr_id
not in self
._nsrs
:
3852 err
= "NSR id %s not found " % nsr_id
3853 self
._log
.error(err
)
3854 raise NetworkServiceRecordError(err
)
3856 nsr
= self
._nsrs
[nsr_id
]
3857 yield from nsr
.nsm_plugin
.instantiate_ns(nsr
, config_xact
)
3860 def update_vnfr(self
, vnfr
):
3861 """Create/Update an VNFR """
3863 vnfr_state
= self
._vnfrs
[vnfr
.id].state
3864 self
._log
.debug("Updating VNFR with state %s: vnfr %s", vnfr_state
, vnfr
)
3866 yield from self
._vnfrs
[vnfr
.id].update_state(vnfr
)
3867 nsr
= self
.find_nsr_for_vnfr(vnfr
.id)
3868 yield from nsr
.update_state()
3870 def find_nsr_for_vnfr(self
, vnfr_id
):
3871 """ Find the NSR which )has the passed vnfr id"""
3872 for nsr
in list(self
.nsrs
.values()):
3873 for vnfr
in list(nsr
.vnfrs
.values()):
3874 if vnfr
.id == vnfr_id
:
3878 def delete_vnfr(self
, vnfr_id
):
3879 """ Delete VNFR with the passed id"""
3880 del self
._vnfrs
[vnfr_id
]
3882 def get_nsd_ref(self
, nsd_id
):
3883 """ Get network service descriptor for the passed nsd_id
3885 nsd
= self
.get_nsd(nsd_id
)
3890 def get_nsr_config(self
, nsd_id
):
3891 xpath
= "C,/nsr:ns-instance-config"
3892 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3894 for result
in results
:
3895 entry
= yield from result
3896 ns_instance_config
= entry
.result
3898 for nsr
in ns_instance_config
.nsr
:
3899 if nsr
.nsd
.id == nsd_id
:
3905 def nsd_unref_by_nsr_id(self
, nsr_id
):
3906 """ Unref the network service descriptor based on NSR id """
3907 self
._log
.debug("NSR Unref called for Nsr Id:%s", nsr_id
)
3908 if nsr_id
in self
._nsrs
:
3909 nsr
= self
._nsrs
[nsr_id
]
3912 nsd
= self
.get_nsd(nsr
.nsd_id
)
3913 self
._log
.debug("Releasing ref on NSD %s held by NSR %s - Curr %d",
3914 nsd
.id, nsr
.id, nsd
.ref_count
)
3916 except NetworkServiceDescriptorError
:
3917 # We store a copy of NSD in NSR and the NSD in nsd-catalog
3922 self
._log
.error("Cannot find NSR with id %s", nsr_id
)
3923 raise NetworkServiceDescriptorUnrefError("No NSR with id" % nsr_id
)
3926 def nsd_unref(self
, nsd_id
):
3927 """ Unref the network service descriptor associated with the id """
3928 nsd
= self
.get_nsd(nsd_id
)
3931 def get_nsd(self
, nsd_id
):
3932 """ Get network service descriptor for the passed nsd_id"""
3933 if nsd_id
not in self
._nsds
:
3934 self
._log
.error("Cannot find NSD id:%s", nsd_id
)
3935 raise NetworkServiceDescriptorError("Cannot find NSD id:%s", nsd_id
)
3937 return self
._nsds
[nsd_id
]
3939 def create_nsd(self
, nsd_msg
):
3940 """ Create a network service descriptor """
3941 self
._log
.debug("Create network service descriptor - %s", nsd_msg
)
3942 if nsd_msg
.id in self
._nsds
:
3943 self
._log
.error("Cannot create NSD %s -NSD ID already exists", nsd_msg
)
3944 raise NetworkServiceDescriptorError("NSD already exists-%s", nsd_msg
.id)
3946 nsd
= NetworkServiceDescriptor(
3953 self
._nsds
[nsd_msg
.id] = nsd
3957 def update_nsd(self
, nsd
):
3958 """ update the Network service descriptor """
3959 self
._log
.debug("Update network service descriptor - %s", nsd
)
3960 if nsd
.id not in self
._nsds
:
3961 self
._log
.debug("No NSD found - creating NSD id = %s", nsd
.id)
3962 self
.create_nsd(nsd
)
3964 self
._log
.debug("Updating NSD id = %s, nsd = %s", nsd
.id, nsd
)
3965 self
._nsds
[nsd
.id].update(nsd
)
3967 def delete_nsd(self
, nsd_id
):
3968 """ Delete the Network service descriptor with the passed id """
3969 self
._log
.debug("Deleting the network service descriptor - %s", nsd_id
)
3970 if nsd_id
not in self
._nsds
:
3971 self
._log
.debug("Delete NSD failed - cannot find nsd-id %s", nsd_id
)
3972 raise NetworkServiceDescriptorNotFound("Cannot find %s", nsd_id
)
3974 if nsd_id
not in self
._nsds
:
3975 self
._log
.debug("Cannot delete NSD id %s reference exists %s",
3977 self
._nsds
[nsd_id
].ref_count
)
3978 raise NetworkServiceDescriptorRefCountExists(
3979 "Cannot delete :%s, ref_count:%s",
3981 self
._nsds
[nsd_id
].ref_count
)
3983 del self
._nsds
[nsd_id
]
3985 def get_vnfd_config(self
, xact
):
3986 vnfd_dts_reg
= self
._vnfd
_dts
_handler
.regh
3987 for cfg
in vnfd_dts_reg
.get_xact_elements(xact
):
3988 if cfg
.id not in self
._vnfds
:
3989 self
.create_vnfd(cfg
)
3991 def get_vnfd(self
, vnfd_id
, xact
):
3992 """ Get virtual network function descriptor for the passed vnfd_id"""
3993 if vnfd_id
not in self
._vnfds
:
3994 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
3995 self
.get_vnfd_config(xact
)
3997 if vnfd_id
not in self
._vnfds
:
3998 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
3999 raise VnfDescriptorError("Cannot find VNFD id:%s", vnfd_id
)
4001 return self
._vnfds
[vnfd_id
]
4003 def create_vnfd(self
, vnfd
):
4004 """ Create a virtual network function descriptor """
4005 self
._log
.debug("Create virtual network function descriptor - %s", vnfd
)
4006 if vnfd
.id in self
._vnfds
:
4007 self
._log
.error("Cannot create VNFD %s -VNFD ID already exists", vnfd
)
4008 raise VnfDescriptorError("VNFD already exists-%s", vnfd
.id)
4010 self
._vnfds
[vnfd
.id] = vnfd
4011 return self
._vnfds
[vnfd
.id]
4013 def update_vnfd(self
, vnfd
):
4014 """ Update the virtual network function descriptor """
4015 self
._log
.debug("Update virtual network function descriptor- %s", vnfd
)
4018 if vnfd
.id not in self
._vnfds
:
4019 self
._log
.debug("No VNFD found - creating VNFD id = %s", vnfd
.id)
4020 self
.create_vnfd(vnfd
)
4022 self
._log
.debug("Updating VNFD id = %s, vnfd = %s", vnfd
.id, vnfd
)
4023 self
._vnfds
[vnfd
.id] = vnfd
4026 def delete_vnfd(self
, vnfd_id
):
4027 """ Delete the virtual network function descriptor with the passed id """
4028 self
._log
.debug("Deleting the virtual network function descriptor - %s", vnfd_id
)
4029 if vnfd_id
not in self
._vnfds
:
4030 self
._log
.debug("Delete VNFD failed - cannot find vnfd-id %s", vnfd_id
)
4031 raise VnfDescriptorError("Cannot find %s", vnfd_id
)
4033 del self
._vnfds
[vnfd_id
]
4035 def nsd_in_use(self
, nsd_id
):
4036 """ Is the NSD with the passed id in use """
4037 self
._log
.debug("Is this NSD in use - msg:%s", nsd_id
)
4038 if nsd_id
in self
._nsds
:
4039 return self
._nsds
[nsd_id
].in_use()
4043 def publish_nsr(self
, xact
, path
, msg
):
4044 """ Publish a NSR """
4045 self
._log
.debug("Publish NSR with path %s, msg %s",
4047 yield from self
.nsr_handler
.update(xact
, path
, msg
)
4050 def unpublish_nsr(self
, xact
, path
):
4051 """ Un Publish an NSR """
4052 self
._log
.debug("Publishing delete NSR with path %s", path
)
4053 yield from self
.nsr_handler
.delete(path
, xact
)
4055 def vnfr_is_ready(self
, vnfr_id
):
4056 """ VNFR with the id is ready """
4057 self
._log
.debug("VNFR id %s ready", vnfr_id
)
4058 if vnfr_id
not in self
._vnfds
:
4059 err
= "Did not find VNFR ID with id %s" % vnfr_id
4060 self
._log
.critical("err")
4061 raise VirtualNetworkFunctionRecordError(err
)
4062 self
._vnfrs
[vnfr_id
].is_ready()
4065 def get_nsd_refcount(self
, nsd_id
):
4066 """ Get the nsd_list from this NSM"""
4068 def nsd_refcount_xpath(nsd_id
):
4069 """ xpath for ref count entry """
4070 return (NsdRefCountDtsHandler
.XPATH
+
4071 "[rw-nsr:nsd-id-ref = '{}']").format(nsd_id
)
4074 if nsd_id
is None or nsd_id
== "":
4075 for nsd
in self
._nsds
.values():
4076 nsd_msg
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4077 nsd_msg
.nsd_id_ref
= nsd
.id
4078 nsd_msg
.instance_ref_count
= nsd
.ref_count
4079 nsd_list
.append((nsd_refcount_xpath(nsd
.id), nsd_msg
))
4080 elif nsd_id
in self
._nsds
:
4081 nsd_msg
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4082 nsd_msg
.nsd_id_ref
= self
._nsds
[nsd_id
].id
4083 nsd_msg
.instance_ref_count
= self
._nsds
[nsd_id
].ref_count
4084 nsd_list
.append((nsd_refcount_xpath(nsd_id
), nsd_msg
))
4089 def terminate_ns(self
, nsr_id
, xact
):
4091 Terminate network service for the given NSR Id
4094 # Terminate the instances/networks assocaited with this nw service
4095 self
._log
.debug("Terminating the network service %s", nsr_id
)
4097 yield from self
._nsrs
[nsr_id
].terminate()
4098 except Exception as e
:
4099 self
.log
.exception("Failed to terminate NSR[id=%s]", nsr_id
)
4102 yield from self
.nsd_unref_by_nsr_id(nsr_id
)
4104 # Unpublish the NSR record
4105 self
._log
.debug("Unpublishing the network service %s", nsr_id
)
4106 yield from self
._nsrs
[nsr_id
].unpublish(xact
)
4108 # Finaly delete the NS instance from this NS Manager
4109 self
._log
.debug("Deletng the network service %s", nsr_id
)
4110 self
.delete_nsr(nsr_id
)
4113 class NsmRecordsPublisherProxy(object):
4114 """ This class provides a publisher interface that allows plugin objects
4115 to publish NSR/VNFR/VLR"""
4117 def __init__(self
, dts
, log
, loop
, nsr_pub_hdlr
, vnfr_pub_hdlr
, vlr_pub_hdlr
):
4121 self
._nsr
_pub
_hdlr
= nsr_pub_hdlr
4122 self
._vlr
_pub
_hdlr
= vlr_pub_hdlr
4123 self
._vnfr
_pub
_hdlr
= vnfr_pub_hdlr
4126 def publish_nsr(self
, xact
, nsr
):
4127 """ Publish an NSR """
4128 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4129 return (yield from self
._nsr
_pub
_hdlr
.update(xact
, path
, nsr
))
4132 def unpublish_nsr(self
, xact
, nsr
):
4133 """ Unpublish an NSR """
4134 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4135 return (yield from self
._nsr
_pub
_hdlr
.delete(xact
, path
))
4138 def publish_vnfr(self
, xact
, vnfr
):
4139 """ Publish an VNFR """
4140 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4141 return (yield from self
._vnfr
_pub
_hdlr
.update(xact
, path
, vnfr
))
4144 def unpublish_vnfr(self
, xact
, vnfr
):
4145 """ Unpublish a VNFR """
4146 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4147 return (yield from self
._vnfr
_pub
_hdlr
.delete(xact
, path
))
4150 def publish_vlr(self
, xact
, vlr
):
4151 """ Publish a VLR """
4152 path
= VirtualLinkRecord
.vlr_xpath(vlr
)
4153 return (yield from self
._vlr
_pub
_hdlr
.update(xact
, path
, vlr
))
4156 def unpublish_vlr(self
, xact
, vlr
):
4157 """ Unpublish a VLR """
4158 path
= VirtualLinkRecord
.vlr_xpath(vlr
)
4159 return (yield from self
._vlr
_pub
_hdlr
.delete(xact
, path
))
4162 class ScalingRpcHandler(mano_dts
.DtsHandler
):
4163 """ The Network service Monitor DTS handler """
4164 SCALE_IN_INPUT_XPATH
= "I,/nsr:exec-scale-in"
4165 SCALE_IN_OUTPUT_XPATH
= "O,/nsr:exec-scale-in"
4167 SCALE_OUT_INPUT_XPATH
= "I,/nsr:exec-scale-out"
4168 SCALE_OUT_OUTPUT_XPATH
= "O,/nsr:exec-scale-out"
4170 ACTION
= Enum('ACTION', 'SCALE_IN SCALE_OUT')
4172 def __init__(self
, log
, dts
, loop
, callback
=None):
4173 super().__init
__(log
, dts
, loop
)
4174 self
.callback
= callback
4175 self
.last_instance_id
= defaultdict(int)
4181 def on_scale_in_prepare(xact_info
, action
, ks_path
, msg
):
4182 assert action
== rwdts
.QueryAction
.RPC
4186 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_IN
)
4188 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleIn
.from_dict({
4189 "instance_id": msg
.instance_id
})
4191 xact_info
.respond_xpath(
4192 rwdts
.XactRspCode
.ACK
,
4193 self
.__class
__.SCALE_IN_OUTPUT_XPATH
,
4196 except Exception as e
:
4197 self
.log
.exception(e
)
4198 xact_info
.respond_xpath(
4199 rwdts
.XactRspCode
.NACK
,
4200 self
.__class
__.SCALE_IN_OUTPUT_XPATH
)
4203 def on_scale_out_prepare(xact_info
, action
, ks_path
, msg
):
4204 assert action
== rwdts
.QueryAction
.RPC
4207 scaling_group
= msg
.scaling_group_name_ref
4208 if not msg
.instance_id
:
4209 last_instance_id
= self
.last_instance_id
[scale_group
]
4210 msg
.instance_id
= last_instance_id
+ 1
4211 self
.last_instance_id
[scale_group
] += 1
4214 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_OUT
)
4216 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleOut
.from_dict({
4217 "instance_id": msg
.instance_id
})
4219 xact_info
.respond_xpath(
4220 rwdts
.XactRspCode
.ACK
,
4221 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
,
4224 except Exception as e
:
4225 self
.log
.exception(e
)
4226 xact_info
.respond_xpath(
4227 rwdts
.XactRspCode
.NACK
,
4228 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
)
4230 scale_in_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4231 on_prepare
=on_scale_in_prepare
)
4232 scale_out_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4233 on_prepare
=on_scale_out_prepare
)
4235 with self
.dts
.group_create() as group
:
4237 xpath
=self
.__class
__.SCALE_IN_INPUT_XPATH
,
4238 handler
=scale_in_hdl
,
4239 flags
=rwdts
.Flag
.PUBLISHER
)
4241 xpath
=self
.__class
__.SCALE_OUT_INPUT_XPATH
,
4242 handler
=scale_out_hdl
,
4243 flags
=rwdts
.Flag
.PUBLISHER
)
4246 class NsmTasklet(rift
.tasklets
.Tasklet
):
4248 The network service manager tasklet
4250 def __init__(self
, *args
, **kwargs
):
4251 super(NsmTasklet
, self
).__init
__(*args
, **kwargs
)
4252 self
.rwlog
.set_category("rw-mano-log")
4253 self
.rwlog
.set_subcategory("nsm")
4258 self
._ro
_plugin
_selector
= None
4259 self
._vnffgmgr
= None
4261 self
._nsr
_handler
= None
4262 self
._vnfr
_pub
_handler
= None
4263 self
._vlr
_pub
_handler
= None
4264 self
._vnfd
_pub
_handler
= None
4265 self
._scale
_cfg
_handler
= None
4267 self
._records
_publisher
_proxy
= None
4270 """ The task start callback """
4271 super(NsmTasklet
, self
).start()
4272 self
.log
.info("Starting NsmTasklet")
4274 self
.log
.debug("Registering with dts")
4275 self
._dts
= rift
.tasklets
.DTS(self
.tasklet_info
,
4276 RwNsmYang
.get_schema(),
4278 self
.on_dts_state_change
)
4280 self
.log
.debug("Created DTS Api GI Object: %s", self
._dts
)
4286 print("Caught Exception in NSM stop:", sys
.exc_info()[0])
4289 def on_instance_started(self
):
4290 """ Task instance started callback """
4291 self
.log
.debug("Got instance started callback")
4295 """ Task init callback """
4296 self
.log
.debug("Got instance started callback")
4298 self
.log
.debug("creating config account handler")
4300 self
._nsr
_pub
_handler
= publisher
.NsrOpDataDtsHandler(self
._dts
, self
.log
, self
.loop
)
4301 yield from self
._nsr
_pub
_handler
.register()
4303 self
._vnfr
_pub
_handler
= publisher
.VnfrPublisherDtsHandler(self
._dts
, self
.log
, self
.loop
)
4304 yield from self
._vnfr
_pub
_handler
.register()
4306 self
._vlr
_pub
_handler
= publisher
.VlrPublisherDtsHandler(self
._dts
, self
.log
, self
.loop
)
4307 yield from self
._vlr
_pub
_handler
.register()
4309 manifest
= self
.tasklet_info
.get_pb_manifest()
4310 use_ssl
= manifest
.bootstrap_phase
.rwsecurity
.use_ssl
4311 ssl_cert
= manifest
.bootstrap_phase
.rwsecurity
.cert
4312 ssl_key
= manifest
.bootstrap_phase
.rwsecurity
.key
4314 self
._vnfd
_pub
_handler
= publisher
.VnfdPublisher(use_ssl
, ssl_cert
, ssl_key
, self
.loop
)
4316 self
._records
_publisher
_proxy
= NsmRecordsPublisherProxy(
4320 self
._nsr
_pub
_handler
,
4321 self
._vnfr
_pub
_handler
,
4322 self
._vlr
_pub
_handler
,
4325 # Register the NSM to receive the nsm plugin
4326 # when cloud account is configured
4327 self
._ro
_plugin
_selector
= cloud
.ROAccountPluginSelector(
4331 self
._records
_publisher
_proxy
,
4333 yield from self
._ro
_plugin
_selector
.register()
4335 self
._cloud
_account
_handler
= cloud
.CloudAccountConfigSubscriber(
4340 yield from self
._cloud
_account
_handler
.register()
4342 self
._vnffgmgr
= rwvnffgmgr
.VnffgMgr(self
._dts
, self
.log
, self
.log_hdl
, self
.loop
)
4343 yield from self
._vnffgmgr
.register()
4345 self
._nsm
= NsManager(
4349 self
._nsr
_pub
_handler
,
4350 self
._vnfr
_pub
_handler
,
4351 self
._vlr
_pub
_handler
,
4352 self
._ro
_plugin
_selector
,
4354 self
._vnfd
_pub
_handler
,
4355 self
._cloud
_account
_handler
4358 yield from self
._nsm
.register()
4362 """ Task run callback """
4366 def on_dts_state_change(self
, state
):
4367 """Take action according to current dts state to transition
4368 application into the corresponding application state
4371 state - current dts state
4374 rwdts
.State
.INIT
: rwdts
.State
.REGN_COMPLETE
,
4375 rwdts
.State
.CONFIG
: rwdts
.State
.RUN
,
4379 rwdts
.State
.INIT
: self
.init
,
4380 rwdts
.State
.RUN
: self
.run
,
4383 # Transition application to next state
4384 handler
= handlers
.get(state
, None)
4385 if handler
is not None:
4386 yield from handler()
4388 # Transition dts to next state
4389 next_state
= switch
.get(state
, None)
4390 if next_state
is not None:
4391 self
.log
.debug("Changing state to %s", next_state
)
4392 self
._dts
.handle
.set_state(next_state
)