2 # Copyright 2016 RIFT.IO Inc
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
17 # vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
20 import ncclient
.asyncio_manager
32 from collections
import deque
33 from collections
import defaultdict
37 gi
.require_version('RwYang', '1.0')
38 gi
.require_version('RwNsdYang', '1.0')
39 gi
.require_version('RwDts', '1.0')
40 gi
.require_version('RwNsmYang', '1.0')
41 gi
.require_version('RwNsrYang', '1.0')
42 gi
.require_version('RwTypes', '1.0')
43 gi
.require_version('RwVlrYang', '1.0')
44 gi
.require_version('RwVnfrYang', '1.0')
45 from gi
.repository
import (
61 import rift
.mano
.ncclient
62 import rift
.mano
.config_data
.config
63 import rift
.mano
.dts
as mano_dts
65 from . import rwnsm_conman
as conman
67 from . import publisher
69 from . import config_value_pool
70 from . import rwvnffgmgr
71 from . import scale_group
74 class NetworkServiceRecordState(Enum
):
75 """ Network Service Record State """
79 VNFFG_INIT_PHASE
= 104
85 VL_TERMINATE_PHASE
= 111
86 VNF_TERMINATE_PHASE
= 112
87 VNFFG_TERMINATE_PHASE
= 113
94 class NetworkServiceRecordError(Exception):
95 """ Network Service Record Error """
99 class NetworkServiceDescriptorError(Exception):
100 """ Network Service Descriptor Error """
104 class VirtualNetworkFunctionRecordError(Exception):
105 """ Virtual Network Function Record Error """
109 class NetworkServiceDescriptorNotFound(Exception):
110 """ Cannot find Network Service Descriptor"""
114 class NetworkServiceDescriptorRefCountExists(Exception):
115 """ Network Service Descriptor reference count exists """
119 class NetworkServiceDescriptorUnrefError(Exception):
120 """ Failed to unref a network service descriptor """
124 class NsrInstantiationFailed(Exception):
125 """ Failed to instantiate network service """
129 class VnfInstantiationFailed(Exception):
130 """ Failed to instantiate virtual network function"""
134 class VnffgInstantiationFailed(Exception):
135 """ Failed to instantiate virtual network function"""
139 class VnfDescriptorError(Exception):
140 """Failed to instantiate virtual network function"""
144 class ScalingOperationError(Exception):
148 class ScaleGroupMissingError(Exception):
152 class PlacementGroupError(Exception):
156 class NsrNsdUpdateError(Exception):
160 class NsrVlUpdateError(NsrNsdUpdateError
):
164 class VlRecordState(Enum
):
165 """ VL Record State """
167 INSTANTIATION_PENDING
= 102
169 TERMINATE_PENDING
= 104
174 class VnffgRecordState(Enum
):
175 """ VNFFG Record State """
177 INSTANTIATION_PENDING
= 102
179 TERMINATE_PENDING
= 104
184 class VnffgRecord(object):
185 """ Vnffg Records class"""
188 def __init__(self
, dts
, log
, loop
, vnffgmgr
, nsr
, nsr_name
, vnffgd_msg
, sdn_account_name
):
193 self
._vnffgmgr
= vnffgmgr
195 self
._nsr
_name
= nsr_name
196 self
._vnffgd
_msg
= vnffgd_msg
197 if sdn_account_name
is None:
198 self
._sdn
_account
_name
= ''
200 self
._sdn
_account
_name
= sdn_account_name
202 self
._vnffgr
_id
= str(uuid
.uuid4())
203 self
._vnffgr
_rsp
_id
= list()
204 self
._vnffgr
_state
= VnffgRecordState
.INIT
209 return self
._vnffgr
_id
213 """ state of this VNF """
214 return self
._vnffgr
_state
216 def fetch_vnffgr(self
):
218 Get VNFFGR message to be published
221 if self
._vnffgr
_state
== VnffgRecordState
.INIT
:
222 vnffgr_dict
= {"id": self
._vnffgr
_id
,
223 "nsd_id": self
._nsr
.nsd_id
,
224 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
225 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
226 "sdn_account": self
._sdn
_account
_name
,
227 "operational_status": 'init',
229 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
230 elif self
._vnffgr
_state
== VnffgRecordState
.TERMINATED
:
231 vnffgr_dict
= {"id": self
._vnffgr
_id
,
232 "nsd_id": self
._nsr
.nsd_id
,
233 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
234 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
235 "sdn_account": self
._sdn
_account
_name
,
236 "operational_status": 'terminated',
238 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
241 vnffgr
= self
._vnffgmgr
.fetch_vnffgr(self
._vnffgr
_id
)
243 self
._log
.exception("Fetching VNFFGR for VNFFG with id %s failed", self
._vnffgr
_id
)
244 self
._vnffgr
_state
= VnffgRecordState
.FAILED
245 vnffgr_dict
= {"id": self
._vnffgr
_id
,
246 "nsd_id": self
._nsr
.nsd_id
,
247 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
248 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
249 "sdn_account": self
._sdn
_account
_name
,
250 "operational_status": 'failed',
252 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
257 def vnffgr_create_msg(self
):
258 """ Virtual Link Record message for Creating VLR in VNS """
259 vnffgr_dict
= {"id": self
._vnffgr
_id
,
260 "nsd_id": self
._nsr
.nsd_id
,
261 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
262 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
263 "sdn_account": self
._sdn
_account
_name
,
265 vnffgr
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
266 for rsp
in self
._vnffgd
_msg
.rsp
:
267 vnffgr_rsp
= vnffgr
.rsp
.add()
268 vnffgr_rsp
.id = str(uuid
.uuid4())
269 vnffgr_rsp
.name
= self
._nsr
.name
+ '.' + rsp
.name
270 self
._vnffgr
_rsp
_id
.append(vnffgr_rsp
.id)
271 vnffgr_rsp
.vnffgd_rsp_id_ref
= rsp
.id
272 vnffgr_rsp
.vnffgd_rsp_name_ref
= rsp
.name
273 for rsp_cp_ref
in rsp
.vnfd_connection_point_ref
:
274 vnfd
= [vnfr
.vnfd
for vnfr
in self
._nsr
.vnfrs
.values() if vnfr
.vnfd
.id == rsp_cp_ref
.vnfd_id_ref
]
275 self
._log
.debug("VNFD message during VNFFG instantiation is %s",vnfd
)
276 if len(vnfd
) > 0 and vnfd
[0].has_field('service_function_type'):
277 self
._log
.debug("Service Function Type for VNFD ID %s is %s",rsp_cp_ref
.vnfd_id_ref
, vnfd
[0].service_function_type
)
279 self
._log
.error("Service Function Type not available for VNFD ID %s; Skipping in chain",rsp_cp_ref
.vnfd_id_ref
)
282 vnfr_cp_ref
= vnffgr_rsp
.vnfr_connection_point_ref
.add()
283 vnfr_cp_ref
.member_vnf_index_ref
= rsp_cp_ref
.member_vnf_index_ref
284 vnfr_cp_ref
.hop_number
= rsp_cp_ref
.order
285 vnfr_cp_ref
.vnfd_id_ref
=rsp_cp_ref
.vnfd_id_ref
286 vnfr_cp_ref
.service_function_type
= vnfd
[0].service_function_type
287 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
288 if (nsr_vnfr
.vnfd
.id == vnfr_cp_ref
.vnfd_id_ref
and
289 nsr_vnfr
.member_vnf_index
== vnfr_cp_ref
.member_vnf_index_ref
):
290 vnfr_cp_ref
.vnfr_id_ref
= nsr_vnfr
.id
291 vnfr_cp_ref
.vnfr_name_ref
= nsr_vnfr
.name
292 vnfr_cp_ref
.vnfr_connection_point_ref
= rsp_cp_ref
.vnfd_connection_point_ref
294 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
295 self
._log
.debug(" Received VNFR is %s", vnfr
)
296 while vnfr
.operational_status
!= 'running':
297 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
298 if vnfr
.operational_status
== 'failed':
299 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
300 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self
.id, vnfr
.id))
301 yield from asyncio
.sleep(2, loop
=self
._loop
)
302 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
303 self
._log
.debug("Received VNFR is %s", vnfr
)
305 vnfr_cp_ref
.connection_point_params
.mgmt_address
= vnfr
.mgmt_interface
.ip_address
306 for cp
in vnfr
.connection_point
:
307 if cp
.name
== vnfr_cp_ref
.vnfr_connection_point_ref
:
308 vnfr_cp_ref
.connection_point_params
.port_id
= cp
.connection_point_id
309 vnfr_cp_ref
.connection_point_params
.name
= self
._nsr
.name
+ '.' + cp
.name
310 for vdu
in vnfr
.vdur
:
311 for ext_intf
in vdu
.external_interface
:
312 if ext_intf
.name
== vnfr_cp_ref
.vnfr_connection_point_ref
:
313 vnfr_cp_ref
.connection_point_params
.vm_id
= vdu
.vim_id
314 self
._log
.debug("VIM ID for CP %s in VNFR %s is %s",cp
.name
,nsr_vnfr
.id,
315 vnfr_cp_ref
.connection_point_params
.vm_id
)
318 vnfr_cp_ref
.connection_point_params
.address
= cp
.ip_address
319 vnfr_cp_ref
.connection_point_params
.port
= VnffgRecord
.SFF_DP_PORT
321 for vnffgd_classifier
in self
._vnffgd
_msg
.classifier
:
322 _rsp
= [rsp
for rsp
in vnffgr
.rsp
if rsp
.vnffgd_rsp_id_ref
== vnffgd_classifier
.rsp_id_ref
]
324 rsp_id_ref
= _rsp
[0].id
325 rsp_name
= _rsp
[0].name
327 self
._log
.error("RSP with ID %s not found during classifier creation for classifier id %s",vnffgd_classifier
.rsp_id_ref
,vnffgd_classifier
.id)
329 vnffgr_classifier
= vnffgr
.classifier
.add()
330 vnffgr_classifier
.id = vnffgd_classifier
.id
331 vnffgr_classifier
.name
= self
._nsr
.name
+ '.' + vnffgd_classifier
.name
332 _rsp
[0].classifier_name
= vnffgr_classifier
.name
333 vnffgr_classifier
.rsp_id_ref
= rsp_id_ref
334 vnffgr_classifier
.rsp_name
= rsp_name
335 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
336 if (nsr_vnfr
.vnfd
.id == vnffgd_classifier
.vnfd_id_ref
and
337 nsr_vnfr
.member_vnf_index
== vnffgd_classifier
.member_vnf_index_ref
):
338 vnffgr_classifier
.vnfr_id_ref
= nsr_vnfr
.id
339 vnffgr_classifier
.vnfr_name_ref
= nsr_vnfr
.name
340 vnffgr_classifier
.vnfr_connection_point_ref
= vnffgd_classifier
.vnfd_connection_point_ref
342 if nsr_vnfr
.vnfd
.service_function_chain
== 'CLASSIFIER':
343 vnffgr_classifier
.sff_name
= nsr_vnfr
.name
345 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
346 self
._log
.debug(" Received VNFR is %s", vnfr
)
347 while vnfr
.operational_status
!= 'running':
348 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
349 if vnfr
.operational_status
== 'failed':
350 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
351 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self
.id, vnfr
.id))
352 yield from asyncio
.sleep(2, loop
=self
._loop
)
353 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
354 self
._log
.debug("Received VNFR is %s", vnfr
)
356 for cp
in vnfr
.connection_point
:
357 if cp
.name
== vnffgr_classifier
.vnfr_connection_point_ref
:
358 vnffgr_classifier
.port_id
= cp
.connection_point_id
359 vnffgr_classifier
.ip_address
= cp
.ip_address
360 for vdu
in vnfr
.vdur
:
361 for ext_intf
in vdu
.external_interface
:
362 if ext_intf
.name
== vnffgr_classifier
.vnfr_connection_point_ref
:
363 vnffgr_classifier
.vm_id
= vdu
.vim_id
364 self
._log
.debug("VIM ID for CP %s in VNFR %s is %s",cp
.name
,nsr_vnfr
.id,
365 vnfr_cp_ref
.connection_point_params
.vm_id
)
368 self
._log
.info("VNFFGR msg to be sent is %s", vnffgr
)
372 def vnffgr_nsr_sff_list(self
):
373 """ SFF List for VNFR """
375 sf_list
= [nsr_vnfr
.name
for nsr_vnfr
in self
._nsr
.vnfrs
.values() if nsr_vnfr
.vnfd
.service_function_chain
== 'SF']
377 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
378 if (nsr_vnfr
.vnfd
.service_function_chain
== 'CLASSIFIER' or nsr_vnfr
.vnfd
.service_function_chain
== 'SFF'):
379 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
380 self
._log
.debug(" Received VNFR is %s", vnfr
)
381 while vnfr
.operational_status
!= 'running':
382 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
383 if vnfr
.operational_status
== 'failed':
384 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
385 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self
.id, vnfr
.id))
386 yield from asyncio
.sleep(2, loop
=self
._loop
)
387 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
388 self
._log
.debug("Received VNFR is %s", vnfr
)
390 sff
= RwsdnYang
.VNFFGSff()
391 sff_list
[nsr_vnfr
.vnfd
.id] = sff
392 sff
.name
= nsr_vnfr
.name
393 sff
.function_type
= nsr_vnfr
.vnfd
.service_function_chain
395 sff
.mgmt_address
= vnfr
.mgmt_interface
.ip_address
396 sff
.mgmt_port
= VnffgRecord
.SFF_MGMT_PORT
397 for cp
in vnfr
.connection_point
:
398 sff_dp
= sff
.dp_endpoints
.add()
399 sff_dp
.name
= self
._nsr
.name
+ '.' + cp
.name
400 sff_dp
.address
= cp
.ip_address
401 sff_dp
.port
= VnffgRecord
.SFF_DP_PORT
402 if nsr_vnfr
.vnfd
.service_function_chain
== 'SFF':
403 for sf_name
in sf_list
:
404 _sf
= sff
.vnfr_list
.add()
405 _sf
.vnfr_name
= sf_name
410 def instantiate(self
):
411 """ Instantiate this VNFFG """
413 self
._log
.info("Instaniating VNFFGR with vnffgd %s",
417 vnffgr_request
= yield from self
.vnffgr_create_msg()
418 vnffg_sff_list
= yield from self
.vnffgr_nsr_sff_list()
421 vnffgr
= self
._vnffgmgr
.create_vnffgr(vnffgr_request
,self
._vnffgd
_msg
.classifier
,vnffg_sff_list
)
422 except Exception as e
:
423 self
._log
.exception("VNFFG instantiation failed: %s", str(e
))
424 self
._vnffgr
_state
= VnffgRecordState
.FAILED
425 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFFGR %s failure" % (self
.id, vnffgr_request
.id))
427 self
._vnffgr
_state
= VnffgRecordState
.INSTANTIATION_PENDING
429 self
._log
.info("Instantiated VNFFGR :%s", vnffgr
)
430 self
._vnffgr
_state
= VnffgRecordState
.ACTIVE
432 self
._log
.info("Invoking update_state to update NSR state for NSR ID: %s", self
._nsr
.id)
433 yield from self
._nsr
.update_state()
435 def vnffgr_in_vnffgrm(self
):
436 """ Is there a VNFR record in VNFM """
437 if (self
._vnffgr
_state
== VnffgRecordState
.ACTIVE
or
438 self
._vnffgr
_state
== VnffgRecordState
.INSTANTIATION_PENDING
or
439 self
._vnffgr
_state
== VnffgRecordState
.FAILED
):
446 """ Terminate this VNFFGR """
447 if not self
.vnffgr_in_vnffgrm():
448 self
._log
.error("Ignoring terminate request for id %s in state %s",
449 self
.id, self
._vnffgr
_state
)
452 self
._log
.info("Terminating VNFFGR id:%s", self
.id)
453 self
._vnffgr
_state
= VnffgRecordState
.TERMINATE_PENDING
455 self
._vnffgmgr
.terminate_vnffgr(self
._vnffgr
_id
)
457 self
._vnffgr
_state
= VnffgRecordState
.TERMINATED
458 self
._log
.debug("Terminated VNFFGR id:%s", self
.id)
461 class VirtualLinkRecord(object):
462 """ Virtual Link Records class"""
465 def create_record(dts
, log
, loop
, nsr_name
, vld_msg
, cloud_account_name
, om_datacenter
, ip_profile
, nsr_id
, restart_mode
=False):
466 """Creates a new VLR object based on the given data.
468 If restart mode is enabled, then we look for existing records in the
469 DTS and create a VLR records using the exiting data(ID)
474 vlr_obj
= VirtualLinkRecord(
487 res_iter
= yield from dts
.query_read(
488 "D,/vlr:vlr-catalog/vlr:vlr",
489 rwdts
.XactFlag
.MERGE
)
492 response
= yield from fut
493 vlr
= response
.result
495 # Check if the record is already present, if so use the ID of
496 # the existing record. Since the name of the record is uniquely
497 # formed we can use it as a search key!
498 if vlr
.name
== vlr_obj
.name
:
499 vlr_obj
.reset_id(vlr
.id)
504 def __init__(self
, dts
, log
, loop
, nsr_name
, vld_msg
, cloud_account_name
, om_datacenter
, ip_profile
, nsr_id
):
508 self
._nsr
_name
= nsr_name
509 self
._vld
_msg
= vld_msg
510 self
._cloud
_account
_name
= cloud_account_name
511 self
._om
_datacenter
_name
= om_datacenter
512 self
._assigned
_subnet
= None
513 self
._nsr
_id
= nsr_id
514 self
._ip
_profile
= ip_profile
515 self
._vlr
_id
= str(uuid
.uuid4())
516 self
._state
= VlRecordState
.INIT
517 self
._prev
_state
= None
518 self
._create
_time
= int(time
.time())
522 """ path for this object """
523 return "D,/vlr:vlr-catalog/vlr:vlr[vlr:id = '{}']".format(self
._vlr
_id
)
532 """ Get NSR name for this VL """
537 """ Virtual Link Desciptor """
541 def assigned_subnet(self
):
542 """ Subnet assigned to this VL"""
543 return self
._assigned
_subnet
548 Get the name for this VLR.
549 VLR name is "nsr name:VLD name"
551 if self
.vld_msg
.vim_network_name
:
552 return self
.vld_msg
.vim_network_name
553 elif self
.vld_msg
.name
== "multisite":
554 # This is a temporary hack to identify manually provisioned inter-site network
555 return self
.vld_msg
.name
557 return self
._nsr
_name
+ "." + self
.vld_msg
.name
560 def cloud_account_name(self
):
561 """ Cloud account that this VLR should be created in """
562 return self
._cloud
_account
_name
565 def om_datacenter_name(self
):
566 """ Datacenter that this VLR should be created in """
567 return self
._om
_datacenter
_name
571 """ Get the VLR path from VLR """
572 return (VirtualLinkRecord
.XPATH
+ "[vlr:id = '{}']").format(vlr
.id)
580 def state(self
, value
):
581 """ VLR set state """
585 def prev_state(self
):
586 """ VLR previous state """
587 return self
._prev
_state
590 def prev_state(self
, value
):
591 """ VLR set previous state """
592 self
._prev
_state
= value
596 """ Virtual Link Record message for Creating VLR in VNS """
597 vld_fields
= ["short_name",
605 vld_copy_dict
= {k
: v
for k
, v
in self
.vld_msg
.as_dict().items()
608 vlr_dict
= {"id": self
._vlr
_id
,
609 "nsr_id_ref": self
._nsr
_id
,
610 "vld_ref": self
.vld_msg
.id,
612 "create_time": self
._create
_time
,
613 "cloud_account": self
.cloud_account_name
,
614 "om_datacenter": self
.om_datacenter_name
,
617 if self
._ip
_profile
and self
._ip
_profile
.has_field('ip_profile_params'):
618 vlr_dict
['ip_profile_params' ] = self
._ip
_profile
.ip_profile_params
.as_dict()
620 vlr_dict
.update(vld_copy_dict
)
621 vlr
= RwVlrYang
.YangData_Vlr_VlrCatalog_Vlr
.from_dict(vlr_dict
)
624 def reset_id(self
, vlr_id
):
625 self
._vlr
_id
= vlr_id
627 def create_nsr_vlr_msg(self
, vnfrs
):
628 """ The VLR message"""
629 nsr_vlr
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vlr()
630 nsr_vlr
.vlr_ref
= self
._vlr
_id
631 nsr_vlr
.assigned_subnet
= self
.assigned_subnet
632 nsr_vlr
.cloud_account
= self
.cloud_account_name
633 nsr_vlr
.om_datacenter
= self
.om_datacenter_name
635 for conn
in self
.vld_msg
.vnfd_connection_point_ref
:
637 if (vnfr
.vnfd
.id == conn
.vnfd_id_ref
and
638 vnfr
.member_vnf_index
== conn
.member_vnf_index_ref
and
639 self
.cloud_account_name
== vnfr
.cloud_account_name
and
640 self
.om_datacenter_name
== vnfr
.om_datacenter_name
):
641 cp_entry
= nsr_vlr
.vnfr_connection_point_ref
.add()
642 cp_entry
.vnfr_id
= vnfr
.id
643 cp_entry
.connection_point
= conn
.vnfd_connection_point_ref
648 def instantiate(self
):
649 """ Instantiate this VL """
651 self
._log
.debug("Instaniating VLR key %s, vld %s",
652 self
.xpath
, self
._vld
_msg
)
654 self
._state
= VlRecordState
.INSTANTIATION_PENDING
655 self
._log
.debug("Executing VL create path:%s msg:%s",
656 self
.xpath
, self
.vlr_msg
)
658 with self
._dts
.transaction(flags
=0) as xact
:
659 block
= xact
.block_create()
660 block
.add_query_create(self
.xpath
, self
.vlr_msg
)
661 self
._log
.debug("Executing VL create path:%s msg:%s",
662 self
.xpath
, self
.vlr_msg
)
663 res_iter
= yield from block
.execute(now
=True)
669 self
._state
= VlRecordState
.FAILED
670 raise NsrInstantiationFailed("Failed NS %s instantiation due to empty response" % self
.id)
672 if vlr
.operational_status
== 'failed':
673 self
._log
.debug("NS Id:%s VL creation failed for vlr id %s", self
.id, vlr
.id)
674 self
._state
= VlRecordState
.FAILED
675 raise NsrInstantiationFailed("Failed VL %s instantiation (%s)" % (vlr
.id, vlr
.operational_status_details
))
677 self
._log
.info("Instantiated VL with xpath %s and vlr:%s",
679 self
._state
= VlRecordState
.ACTIVE
680 self
._assigned
_subnet
= vlr
.assigned_subnet
682 def vlr_in_vns(self
):
683 """ Is there a VLR record in VNS """
684 if (self
._state
== VlRecordState
.ACTIVE
or
685 self
._state
== VlRecordState
.INSTANTIATION_PENDING
or
686 self
._state
== VlRecordState
.TERMINATE_PENDING
or
687 self
._state
== VlRecordState
.FAILED
):
694 """ Terminate this VL """
695 if not self
.vlr_in_vns():
696 self
._log
.debug("Ignoring terminate request for id %s in state %s",
697 self
.id, self
._state
)
700 self
._log
.debug("Terminating VL id:%s", self
.id)
701 self
._state
= VlRecordState
.TERMINATE_PENDING
703 with self
._dts
.transaction(flags
=0) as xact
:
704 block
= xact
.block_create()
705 block
.add_query_delete(self
.xpath
)
706 yield from block
.execute(flags
=0, now
=True)
708 self
._state
= VlRecordState
.TERMINATED
709 self
._log
.debug("Terminated VL id:%s", self
.id)
712 class VnfRecordState(Enum
):
713 """ Vnf Record State """
715 INSTANTIATION_PENDING
= 102
717 TERMINATE_PENDING
= 104
722 class VirtualNetworkFunctionRecord(object):
723 """ Virtual Network Function Record class"""
724 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
728 def create_record(dts
, log
, loop
, vnfd
, const_vnfd_msg
, nsd_id
, nsr_name
,
729 cloud_account_name
, om_datacenter_name
, nsr_id
, group_name
, group_instance_id
,
730 placement_groups
, restart_mode
=False):
731 """Creates a new VNFR object based on the given data.
733 If restart mode is enabled, then we look for existing records in the
734 DTS and create a VNFR records using the exiting data(ID)
737 VirtualNetworkFunctionRecord
739 vnfr_obj
= VirtualNetworkFunctionRecord(
753 restart_mode
=restart_mode
)
756 res_iter
= yield from dts
.query_read(
757 "D,/vnfr:vnfr-catalog/vnfr:vnfr",
758 rwdts
.XactFlag
.MERGE
)
761 response
= yield from fut
762 vnfr
= response
.result
764 if vnfr
.name
== vnfr_obj
.name
:
765 vnfr_obj
.reset_id(vnfr
.id)
782 group_instance_id
=None,
783 placement_groups
= [],
784 restart_mode
= False):
789 self
._const
_vnfd
_msg
= const_vnfd_msg
790 self
._nsd
_id
= nsd_id
791 self
._nsr
_name
= nsr_name
792 self
._nsr
_id
= nsr_id
793 self
._cloud
_account
_name
= cloud_account_name
794 self
._om
_datacenter
_name
= om_datacenter_name
795 self
._group
_name
= group_name
796 self
._group
_instance
_id
= group_instance_id
797 self
._placement
_groups
= placement_groups
798 self
._config
_status
= NsrYang
.ConfigStates
.INIT
800 self
._prev
_state
= VnfRecordState
.INIT
801 self
._state
= VnfRecordState
.INIT
802 self
._state
_failed
_reason
= None
804 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
807 self
._vnfr
_id
= str(uuid
.uuid4())
809 self
._vnfr
_msg
= self
.create_vnfr_msg()
810 self
._log
.debug("Set VNFR {} config type to {}".
811 format(self
.name
, self
.config_type
))
812 self
.restart_mode
= restart_mode
815 if group_name
is None and group_instance_id
is not None:
816 raise ValueError("Group instance id must not be provided with an empty group name")
826 return "D,/vnfr:vnfr-catalog/vnfr:vnfr[vnfr:id = '{}']".format(self
.id)
831 return self
._vnfr
_msg
834 def const_vnfr_msg(self
):
836 return RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_ConstituentVnfrRef(vnfr_id
=self
.id,cloud_account
=self
.cloud_account_name
,om_datacenter
=self
._om
_datacenter
_name
)
844 def cloud_account_name(self
):
845 """ Cloud account that this VNF should be created in """
846 return self
._cloud
_account
_name
849 def om_datacenter_name(self
):
850 """ Datacenter that this VNF should be created in """
851 return self
._om
_datacenter
_name
856 """ Is this VNF actve """
857 return True if self
._state
== VnfRecordState
.ACTIVE
else False
861 """ state of this VNF """
865 def state_failed_reason(self
):
866 """ Error message in case this VNF is in failed state """
867 return self
._state
_failed
_reason
870 def member_vnf_index(self
):
871 """ Member VNF index """
872 return self
._const
_vnfd
_msg
.member_vnf_index
877 return self
._nsr
_name
881 """ Name of this VNFR """
882 if self
._name
is not None:
885 name_tags
= [self
._nsr
_name
]
887 if self
._group
_name
is not None:
888 name_tags
.append(self
._group
_name
)
890 if self
._group
_instance
_id
is not None:
891 name_tags
.append(str(self
._group
_instance
_id
))
893 name_tags
.extend([self
.vnfd
.name
, str(self
.member_vnf_index
)])
895 self
._name
= "__".join(name_tags
)
900 def vnfr_xpath(vnfr
):
901 """ Get the VNFR path from VNFR """
902 return (VirtualNetworkFunctionRecord
.XPATH
+ "[vnfr:id = '{}']").format(vnfr
.id)
905 def config_type(self
):
906 cfg_types
= ['netconf', 'juju', 'script']
907 for method
in cfg_types
:
908 if self
._vnfd
.vnf_configuration
.has_field(method
):
913 def config_status(self
):
914 """Return the config status as YANG ENUM string"""
915 self
._log
.debug("Map VNFR {} config status {} ({})".
916 format(self
.name
, self
._config
_status
, self
.config_type
))
917 if self
.config_type
== 'none':
918 return 'config_not_needed'
919 elif self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
921 elif self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
926 def set_state(self
, state
):
927 """ set the state of this object """
928 self
._prev
_state
= self
._state
931 def reset_id(self
, vnfr_id
):
932 self
._vnfr
_id
= vnfr_id
933 self
._vnfr
_msg
= self
.create_vnfr_msg()
936 self
.config_store
.merge_vnfd_config(
939 self
.member_vnf_index
,
942 def create_vnfr_msg(self
):
943 """ VNFR message for this VNFR """
951 vnfd_copy_dict
= {k
: v
for k
, v
in self
._vnfd
.as_dict().items() if k
in vnfd_fields
}
954 "nsr_id_ref": self
._nsr
_id
,
955 "vnfd_ref": self
.vnfd
.id,
957 "cloud_account": self
._cloud
_account
_name
,
958 "om_datacenter": self
._om
_datacenter
_name
,
959 "config_status": self
.config_status
961 vnfr_dict
.update(vnfd_copy_dict
)
963 vnfr
= RwVnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr
.from_dict(vnfr_dict
)
964 vnfr
.member_vnf_index_ref
= self
.member_vnf_index
965 vnfr
.vnf_configuration
.from_dict(self
._vnfd
.vnf_configuration
.as_dict())
967 if self
._vnfd
.mgmt_interface
.has_field("port"):
968 vnfr
.mgmt_interface
.port
= self
._vnfd
.mgmt_interface
.port
970 for group_info
in self
._placement
_groups
:
971 group
= vnfr
.placement_groups_info
.add()
972 group
.from_dict(group_info
.as_dict())
974 # UI expects the monitoring param field to exist
975 vnfr
.monitoring_param
= []
977 self
._log
.debug("Get vnfr_msg for VNFR {} : {}".format(self
.name
, vnfr
))
981 def update_vnfm(self
):
982 self
._log
.debug("Send an update to VNFM for VNFR {} with {}".
983 format(self
.name
, self
.vnfr_msg
))
984 yield from self
._dts
.query_update(
986 rwdts
.XactFlag
.TRACE
,
990 def get_config_status(self
):
991 """Return the config status as YANG ENUM"""
992 return self
._config
_status
995 def set_config_status(self
, status
):
997 def status_to_string(status
):
999 NsrYang
.ConfigStates
.INIT
: 'init',
1000 NsrYang
.ConfigStates
.CONFIGURING
: 'configuring',
1001 NsrYang
.ConfigStates
.CONFIG_NOT_NEEDED
: 'config_not_needed',
1002 NsrYang
.ConfigStates
.CONFIGURED
: 'configured',
1003 NsrYang
.ConfigStates
.FAILED
: 'failed',
1006 return status_dc
[status
]
1008 self
._log
.debug("Update VNFR {} from {} ({}) to {}".
1009 format(self
.name
, self
._config
_status
,
1010 self
.config_type
, status
))
1011 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1012 self
._log
.error("Updating already configured VNFR {}".
1016 if self
._config
_status
!= status
:
1018 self
._config
_status
= status
1019 # I don't think this is used. Original implementor can check.
1020 # Caused Exception, so corrected it by status_to_string
1021 # But not sure whats the use of this variable?
1022 self
.vnfr_msg
.config_status
= status_to_string(status
)
1023 except Exception as e
:
1024 self
._log
.error("Exception=%s", str(e
))
1027 self
._log
.debug("Updated VNFR {} status to {}".format(self
.name
, status
))
1029 if self
._config
_status
!= NsrYang
.ConfigStates
.INIT
:
1031 # Publish only after VNFM has the VNFR created
1032 yield from self
.update_vnfm()
1033 except Exception as e
:
1034 self
._log
.error("Exception updating VNFM with new status {} of VNFR {}: {}".
1035 format(status
, self
.name
, e
))
1036 self
._log
.exception(e
)
1038 def is_configured(self
):
1039 if self
.config_type
== 'none':
1042 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1048 def instantiate(self
, nsr
):
1049 """ Instantiate this VNFR"""
1051 self
._log
.debug("Instaniating VNFR key %s, vnfd %s",
1052 self
.xpath
, self
._vnfd
)
1054 self
._log
.debug("Create VNF with xpath %s and vnfr %s",
1055 self
.xpath
, self
.vnfr_msg
)
1057 self
.set_state(VnfRecordState
.INSTANTIATION_PENDING
)
1059 def find_vlr_for_cp(conn
):
1060 """ Find VLR for the given connection point """
1061 for vlr
in nsr
.vlrs
:
1062 for vnfd_cp
in vlr
.vld_msg
.vnfd_connection_point_ref
:
1063 if (vnfd_cp
.vnfd_id_ref
== self
._vnfd
.id and
1064 vnfd_cp
.vnfd_connection_point_ref
== conn
.name
and
1065 vnfd_cp
.member_vnf_index_ref
== self
.member_vnf_index
and
1066 vlr
.cloud_account_name
== self
.cloud_account_name
):
1067 self
._log
.debug("Found VLR for cp_name:%s and vnf-index:%d",
1068 conn
.name
, self
.member_vnf_index
)
1072 # For every connection point in the VNFD fill in the identifier
1073 for conn_p
in self
._vnfd
.connection_point
:
1074 cpr
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_ConnectionPoint()
1075 cpr
.name
= conn_p
.name
1076 cpr
.type_yang
= conn_p
.type_yang
1077 vlr_ref
= find_vlr_for_cp(conn_p
)
1079 msg
= "Failed to find VLR for cp = %s" % conn_p
.name
1080 self
._log
.debug("%s", msg
)
1081 # raise VirtualNetworkFunctionRecordError(msg)
1084 cpr
.vlr_ref
= vlr_ref
.id
1085 self
.vnfr_msg
.connection_point
.append(cpr
)
1086 self
._log
.debug("Connection point [%s] added, vnf id=%s vnfd id=%s",
1087 cpr
, self
.vnfr_msg
.id, self
.vnfr_msg
.vnfd_ref
)
1089 if not self
.restart_mode
:
1090 yield from self
._dts
.query_create(self
.xpath
,
1094 yield from self
._dts
.query_update(self
.xpath
,
1098 self
._log
.info("Created VNF with xpath %s and vnfr %s",
1099 self
.xpath
, self
.vnfr_msg
)
1101 self
._log
.info("Instantiated VNFR with xpath %s and vnfd %s, vnfr %s",
1102 self
.xpath
, self
._vnfd
, self
.vnfr_msg
)
1105 def update_state(self
, vnfr_msg
):
1106 """ Update this VNFR"""
1107 if vnfr_msg
.operational_status
== "running":
1108 if self
.vnfr_msg
.operational_status
!= "running":
1109 yield from self
.is_active()
1110 elif vnfr_msg
.operational_status
== "failed":
1111 yield from self
.instantiation_failed(failed_reason
=vnfr_msg
.operational_status_details
)
1114 def is_active(self
):
1115 """ This VNFR is active """
1116 self
._log
.debug("VNFR %s is active", self
._vnfr
_id
)
1117 self
.set_state(VnfRecordState
.ACTIVE
)
1120 def instantiation_failed(self
, failed_reason
=None):
1121 """ This VNFR instantiation failed"""
1122 self
._log
.error("VNFR %s instantiation failed", self
._vnfr
_id
)
1123 self
.set_state(VnfRecordState
.FAILED
)
1124 self
._state
_failed
_reason
= failed_reason
1126 def vnfr_in_vnfm(self
):
1127 """ Is there a VNFR record in VNFM """
1128 if (self
._state
== VnfRecordState
.ACTIVE
or
1129 self
._state
== VnfRecordState
.INSTANTIATION_PENDING
or
1130 self
._state
== VnfRecordState
.FAILED
):
1136 def terminate(self
):
1137 """ Terminate this VNF """
1138 if not self
.vnfr_in_vnfm():
1139 self
._log
.debug("Ignoring terminate request for id %s in state %s",
1140 self
.id, self
._state
)
1143 self
._log
.debug("Terminating VNF id:%s", self
.id)
1144 self
.set_state(VnfRecordState
.TERMINATE_PENDING
)
1145 with self
._dts
.transaction(flags
=0) as xact
:
1146 block
= xact
.block_create()
1147 block
.add_query_delete(self
.xpath
)
1148 yield from block
.execute(flags
=0)
1149 self
.set_state(VnfRecordState
.TERMINATED
)
1150 self
._log
.debug("Terminated VNF id:%s", self
.id)
1153 class NetworkServiceStatus(object):
1154 """ A class representing the Network service's status """
1155 MAX_EVENTS_RECORDED
= 10
1156 """ Network service Status class"""
1157 def __init__(self
, dts
, log
, loop
):
1162 self
._state
= NetworkServiceRecordState
.INIT
1163 self
._events
= deque([])
1166 def create_notification(self
, evt
, evt_desc
, evt_details
):
1167 xp
= "N,/rw-nsr:nsm-notification"
1168 notif
= RwNsrYang
.YangNotif_RwNsr_NsmNotification()
1170 notif
.description
= evt_desc
1171 notif
.details
= evt_details
if evt_details
is not None else None
1173 yield from self
._dts
.query_create(xp
, rwdts
.XactFlag
.ADVISE
, notif
)
1174 self
._log
.info("Notification called by creating dts query: %s", notif
)
1176 def record_event(self
, evt
, evt_desc
, evt_details
):
1177 """ Record an event """
1178 self
._log
.debug("Recording event - evt %s, evt_descr %s len = %s",
1179 evt
, evt_desc
, len(self
._events
))
1180 if len(self
._events
) >= NetworkServiceStatus
.MAX_EVENTS_RECORDED
:
1181 self
._events
.popleft()
1182 self
._events
.append((int(time
.time()), evt
, evt_desc
,
1183 evt_details
if evt_details
is not None else None))
1185 self
._loop
.create_task(self
.create_notification(evt
,evt_desc
,evt_details
))
1187 def set_state(self
, state
):
1188 """ set the state of this status object """
1192 """ Return the state as a yang enum string """
1193 state_to_str_map
= {"INIT": "init",
1194 "VL_INIT_PHASE": "vl_init_phase",
1195 "VNF_INIT_PHASE": "vnf_init_phase",
1196 "VNFFG_INIT_PHASE": "vnffg_init_phase",
1197 "SCALING_GROUP_INIT_PHASE": "scaling_group_init_phase",
1198 "RUNNING": "running",
1199 "SCALING_OUT": "scaling_out",
1200 "SCALING_IN": "scaling_in",
1201 "TERMINATE_RCVD": "terminate_rcvd",
1202 "TERMINATE": "terminate",
1203 "VL_TERMINATE_PHASE": "vl_terminate_phase",
1204 "VNF_TERMINATE_PHASE": "vnf_terminate_phase",
1205 "VNFFG_TERMINATE_PHASE": "vnffg_terminate_phase",
1206 "TERMINATED": "terminated",
1208 "VL_INSTANTIATE": "vl_instantiate",
1209 "VL_TERMINATE": "vl_terminate",
1211 return state_to_str_map
[self
._state
.name
]
1215 """ State of this status object """
1220 """ Network Service Record as a message"""
1223 for entry
in self
._events
:
1224 event
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_OperationalEvents()
1227 event
.timestamp
, event
.event
, event
.description
, event
.details
= entry
1228 event_list
.append(event
)
1232 class NetworkServiceRecord(object):
1233 """ Network service record """
1234 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
1236 def __init__(self
, dts
, log
, loop
, nsm
, nsm_plugin
, nsr_cfg_msg
, sdn_account_name
, key_pairs
, restart_mode
=False):
1241 self
._nsr
_cfg
_msg
= nsr_cfg_msg
1242 self
._nsm
_plugin
= nsm_plugin
1243 self
._sdn
_account
_name
= sdn_account_name
1246 self
._nsr
_msg
= None
1247 self
._nsr
_regh
= None
1248 self
._key
_pairs
= key_pairs
1253 self
._param
_pools
= {}
1254 self
._scaling
_groups
= {}
1255 self
._create
_time
= int(time
.time())
1256 self
._op
_status
= NetworkServiceStatus(dts
, log
, loop
)
1257 self
._config
_status
= NsrYang
.ConfigStates
.CONFIGURING
1258 self
._config
_status
_details
= None
1260 self
.restart_mode
= restart_mode
1261 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
1262 self
._debug
_running
= False
1263 self
._is
_active
= False
1264 self
._vl
_phase
_completed
= False
1265 self
._vnf
_phase
_completed
= False
1268 # Initalise the state to init
1269 # The NSR moves through the following transitions
1270 # 1. INIT -> VLS_READY once all the VLs in the NSD are created
1271 # 2. VLS_READY - VNFS_READY when all the VNFs in the NSD are created
1272 # 3. VNFS_READY - READY when the NSR is published
1274 self
.set_state(NetworkServiceRecordState
.INIT
)
1276 self
.substitute_input_parameters
= InputParameterSubstitution(self
._log
)
1279 def nsm_plugin(self
):
1281 return self
._nsm
_plugin
1283 def set_state(self
, state
):
1284 """ Set state for this NSR"""
1285 self
._log
.debug("Setting state to %s", state
)
1286 # We are in init phase and is moving to the next state
1287 # The new state could be a FAILED state or VNF_INIIT_PHASE
1288 if self
.state
== NetworkServiceRecordState
.VL_INIT_PHASE
:
1289 self
._vl
_phase
_completed
= True
1291 if self
.state
== NetworkServiceRecordState
.VNF_INIT_PHASE
:
1292 self
._vnf
_phase
_completed
= True
1294 self
._op
_status
.set_state(state
)
1298 """ Get id for this NSR"""
1299 return self
._nsr
_cfg
_msg
.id
1303 """ Name of this network service record """
1304 return self
._nsr
_cfg
_msg
.name
1307 def cloud_account_name(self
):
1308 return self
._nsr
_cfg
_msg
.cloud_account
1311 def om_datacenter_name(self
):
1312 if self
._nsr
_cfg
_msg
.has_field('om_datacenter'):
1313 return self
._nsr
_cfg
_msg
.om_datacenter
1318 """State of this NetworkServiceRecord"""
1319 return self
._op
_status
.state
1323 """ Is this NSR active ?"""
1324 return True if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
else False
1328 """ VLRs associated with this NSR"""
1333 """ VNFRs associated with this NSR"""
1338 """ VNFFGRs associated with this NSR"""
1339 return self
._vnffgrs
1342 def scaling_groups(self
):
1343 """ Scaling groups associated with this NSR """
1344 return self
._scaling
_groups
1347 def param_pools(self
):
1348 """ Parameter value pools associated with this NSR"""
1349 return self
._param
_pools
1352 def nsr_cfg_msg(self
):
1353 return self
._nsr
_cfg
_msg
1356 def nsr_cfg_msg(self
, msg
):
1357 self
._nsr
_cfg
_msg
= msg
1361 """ NSD Protobuf for this NSR """
1362 if self
._nsd
is not None:
1364 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
1369 """ NSD ID for this NSR """
1370 return self
.nsd_msg
.id
1374 ''' Get a new job id for config primitive'''
1379 def config_status(self
):
1380 """ Config status for NSR """
1381 return self
._config
_status
1383 def resolve_placement_group_cloud_construct(self
, input_group
):
1385 Returns the cloud specific construct for placement group
1387 copy_dict
= ['name', 'requirement', 'strategy']
1389 for group_info
in self
._nsr
_cfg
_msg
.nsd_placement_group_maps
:
1390 if group_info
.placement_group_ref
== input_group
.name
:
1391 group
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_PlacementGroupsInfo()
1392 group_dict
= {k
:v
for k
,v
in
1393 group_info
.as_dict().items() if k
!= 'placement_group_ref'}
1394 for param
in copy_dict
:
1395 group_dict
.update({param
: getattr(input_group
, param
)})
1396 group
.from_dict(group_dict
)
1402 return "NSR(name={}, nsd_id={}, cloud_account={})".format(
1403 self
.name
, self
.nsd_id
, self
.cloud_account_name
1406 def _get_vnfd(self
, vnfd_id
, config_xact
):
1407 """ Fetch vnfd msg for the passed vnfd id """
1408 return self
._nsm
.get_vnfd(vnfd_id
, config_xact
)
1410 def _get_vnfd_cloud_account(self
, vnfd_member_index
):
1411 """ Fetch Cloud Account for the passed vnfd id """
1412 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1413 vim_accounts
= [(vnf
.cloud_account
,vnf
.om_datacenter
) for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map \
1414 if vnfd_member_index
== vnf
.member_vnf_index_ref
]
1415 if vim_accounts
and vim_accounts
[0]:
1416 return vim_accounts
[0]
1417 return (self
.cloud_account_name
,self
.om_datacenter_name
)
1419 def _get_constituent_vnfd_msg(self
, vnf_index
):
1420 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1421 if const_vnfd
.member_vnf_index
== vnf_index
:
1424 raise ValueError("Constituent VNF index %s not found" % vnf_index
)
1426 def record_event(self
, evt
, evt_desc
, evt_details
=None, state
=None):
1427 """ Record an event """
1428 self
._op
_status
.record_event(evt
, evt_desc
, evt_details
)
1429 if state
is not None:
1430 self
.set_state(state
)
1432 def scaling_trigger_str(self
, trigger
):
1433 SCALING_TRIGGER_STRS
= {
1434 NsdYang
.ScalingTrigger
.PRE_SCALE_IN
: 'pre-scale-in',
1435 NsdYang
.ScalingTrigger
.POST_SCALE_IN
: 'post-scale-in',
1436 NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
: 'pre-scale-out',
1437 NsdYang
.ScalingTrigger
.POST_SCALE_OUT
: 'post-scale-out',
1440 return SCALING_TRIGGER_STRS
[trigger
]
1441 except Exception as e
:
1442 self
._log
.error("Scaling trigger mapping error for {} : {}".
1444 self
._log
.exception(e
)
1445 return "Unknown trigger"
1448 def instantiate_vls(self
):
1450 This function instantiates VLs for every VL in this Network Service
1452 self
._log
.debug("Instantiating %d VLs in NSD id %s", len(self
._vlrs
),
1454 for vlr
in self
._vlrs
:
1455 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1456 vlr
.state
= VlRecordState
.ACTIVE
1459 def create(self
, config_xact
):
1460 """ Create this network service"""
1461 # Create virtual links for all the external vnf
1462 # connection points in this NS
1463 yield from self
.create_vls()
1465 # Create VNFs in this network service
1466 yield from self
.create_vnfs(config_xact
)
1468 # Create VNFFG for network service
1469 self
.create_vnffgs()
1471 # Create Scaling Groups for each scaling group in NSD
1472 self
.create_scaling_groups()
1474 # Create Parameter Pools
1475 self
.create_param_pools()
1478 def apply_scale_group_config_script(self
, script
, group
, scale_instance
, trigger
, vnfrs
=None):
1479 """ Apply config based on script for scale group """
1482 def add_vnfrs_data(vnfrs_list
):
1483 """ Add as a dict each of the VNFRs data """
1485 for vnfr
in vnfrs_list
:
1486 self
._log
.debug("Add VNFR {} data".format(vnfr
))
1488 vnfr_data
['name'] = vnfr
.name
1489 if trigger
in [NsdYang
.ScalingTrigger
.PRE_SCALE_IN
, NsdYang
.ScalingTrigger
.POST_SCALE_OUT
]:
1490 # Get VNF management and other IPs, etc
1491 opdata
= yield from self
.fetch_vnfr(vnfr
.xpath
)
1492 self
._log
.debug("VNFR {} op data: {}".format(vnfr
.name
, opdata
))
1494 vnfr_data
['rw_mgmt_ip'] = opdata
.mgmt_interface
.ip_address
1495 vnfr_data
['rw_mgmt_port'] = opdata
.mgmt_interface
.port
1496 except Exception as e
:
1497 self
._log
.error("Unable to get management IP for vnfr {}:{}".
1498 format(vnfr
.name
, e
))
1501 vnfr_data
['connection_points'] = []
1502 for cp
in opdata
.connection_point
:
1504 con_pt
['name'] = cp
.name
1505 con_pt
['ip_address'] = cp
.ip_address
1506 vnfr_data
['connection_points'].append(con_pt
)
1507 except Exception as e
:
1508 self
._log
.error("Exception getting connections points for VNFR {}: {}".
1509 format(vnfr
.name
, e
))
1511 vnfrs_data
.append(vnfr_data
)
1512 self
._log
.debug("VNFRs data: {}".format(vnfrs_data
))
1516 def add_nsr_data(nsr
):
1518 nsr_data
['name'] = nsr
.name
1521 if script
is None or len(script
) == 0:
1522 self
._log
.error("Script not provided for scale group config: {}".format(group
.name
))
1525 if script
[0] == '/':
1528 path
= os
.path
.join(os
.environ
['RIFT_INSTALL'], "usr/bin", script
)
1529 if not os
.path
.exists(path
):
1530 self
._log
.error("Config faled for scale group {}: Script does not exist at {}".
1531 format(group
.name
, path
))
1534 # Build a YAML file with all parameters for the script to execute
1535 # The data consists of 5 sections
1537 # 2. Scale group config
1538 # 3. VNFRs in the scale group
1539 # 4. VNFRs outside scale group
1542 data
['trigger'] = group
.trigger_map(trigger
)
1543 data
['config'] = group
.group_msg
.as_dict()
1546 data
["vnfrs_in_group"] = yield from add_vnfrs_data(vnfrs
)
1548 data
["vnfrs_in_group"] = yield from add_vnfrs_data(scale_instance
.vnfrs
)
1550 data
["vnfrs_others"] = yield from add_vnfrs_data(self
.vnfrs
.values())
1551 data
["nsr"] = add_nsr_data(self
)
1554 with tempfile
.NamedTemporaryFile(delete
=False) as tmp_file
:
1555 tmp_file
.write(yaml
.dump(data
, default_flow_style
=True)
1558 self
._log
.debug("Creating a temp file: {} with input data: {}".
1559 format(tmp_file
.name
, data
))
1561 cmd
= "{} {}".format(path
, tmp_file
.name
)
1562 self
._log
.debug("Running the CMD: {}".format(cmd
))
1563 proc
= yield from asyncio
.create_subprocess_shell(cmd
, loop
=self
._loop
)
1564 rc
= yield from proc
.wait()
1566 self
._log
.error("The script {} for scale group {} config returned: {}".
1567 format(script
, group
.name
, rc
))
1575 def apply_scaling_group_config(self
, trigger
, group
, scale_instance
, vnfrs
=None):
1576 """ Apply the config for the scaling group based on trigger """
1577 if group
is None or scale_instance
is None:
1581 def update_config_status(success
=True, err_msg
=None):
1582 self
._log
.debug("Update %s config status to %r : %s",
1583 scale_instance
, success
, err_msg
)
1584 if (scale_instance
.config_status
== "failed"):
1585 # Do not update the config status if it is already in failed state
1588 if scale_instance
.config_status
== "configured":
1589 # Update only to failed state an already configured scale instance
1591 scale_instance
.config_status
= "failed"
1592 scale_instance
.config_err_msg
= err_msg
1593 yield from self
.update_state()
1595 # We are in configuring state
1596 # Only after post scale out mark instance as configured
1597 if trigger
== NsdYang
.ScalingTrigger
.POST_SCALE_OUT
:
1599 scale_instance
.config_status
= "configured"
1601 scale_instance
.config_status
= "failed"
1602 scale_instance
.config_err_msg
= err_msg
1603 yield from self
.update_state()
1605 config
= group
.trigger_config(trigger
)
1609 self
._log
.debug("Scaling group {} config: {}".format(group
.name
, config
))
1610 if config
.has_field("ns_config_primitive_name_ref"):
1611 config_name
= config
.ns_config_primitive_name_ref
1612 nsd_msg
= self
.nsd_msg
1613 config_primitive
= None
1614 for ns_cfg_prim
in nsd_msg
.service_primitive
:
1615 if ns_cfg_prim
.name
== config_name
:
1616 config_primitive
= ns_cfg_prim
1619 if config_primitive
is None:
1620 raise ValueError("Could not find ns_cfg_prim %s in nsr %s" % (config_name
, self
.name
))
1622 self
._log
.debug("Scaling group {} config primitive: {}".format(group
.name
, config_primitive
))
1623 if config_primitive
.has_field("user_defined_script"):
1624 rc
= yield from self
.apply_scale_group_config_script(config_primitive
.user_defined_script
,
1625 group
, scale_instance
, trigger
, vnfrs
)
1628 err_msg
= "Failed config for trigger {} using config script '{}'". \
1629 format(self
.scaling_trigger_str(trigger
),
1630 config_primitive
.user_defined_script
)
1631 yield from update_config_status(success
=rc
, err_msg
=err_msg
)
1634 err_msg
= "Failed config for trigger {} as config script is not specified". \
1635 format(self
.scaling_trigger_str(trigger
))
1636 yield from update_config_status(success
=False, err_msg
=err_msg
)
1637 raise NotImplementedError("Only script based config support for scale group for now: {}".
1640 err_msg
= "Failed config for trigger {} as config primitive is not specified".\
1641 format(self
.scaling_trigger_str(trigger
))
1642 yield from update_config_status(success
=False, err_msg
=err_msg
)
1643 self
._log
.error("Config primitive not specified for config action in scale group %s" %
1647 def create_scaling_groups(self
):
1648 """ This function creates a NSScalingGroup for every scaling
1649 group defined in he NSD"""
1651 for scaling_group_msg
in self
.nsd_msg
.scaling_group_descriptor
:
1652 self
._log
.debug("Found scaling_group %s in nsr id %s",
1653 scaling_group_msg
.name
, self
.id)
1655 group_record
= scale_group
.ScalingGroup(
1660 self
._scaling
_groups
[group_record
.name
] = group_record
1663 def create_scale_group_instance(self
, group_name
, index
, config_xact
, is_default
=False):
1664 group
= self
._scaling
_groups
[group_name
]
1665 scale_instance
= group
.create_instance(index
, is_default
)
1669 self
._log
.debug("Creating %u VNFs associated with NS id %s scaling group %s",
1670 len(self
.nsd_msg
.constituent_vnfd
), self
.id, self
)
1673 for vnf_index
, count
in group
.vnf_index_count_map
.items():
1674 const_vnfd_msg
= self
._get
_constituent
_vnfd
_msg
(vnf_index
)
1675 vnfd_msg
= self
._get
_vnfd
(const_vnfd_msg
.vnfd_id_ref
, config_xact
)
1677 cloud_account_name
, om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd_msg
.member_vnf_index
)
1678 if cloud_account_name
is None:
1679 cloud_account_name
= self
.cloud_account_name
1680 for _
in range(count
):
1681 vnfr
= yield from self
.create_vnf_record(vnfd_msg
, const_vnfd_msg
, cloud_account_name
, om_datacenter_name
, group_name
, index
)
1682 scale_instance
.add_vnfr(vnfr
)
1688 def instantiate_instance():
1689 self
._log
.debug("Creating %s VNFRS", scale_instance
)
1690 vnfrs
= yield from create_vnfs()
1691 yield from self
.publish()
1693 self
._log
.debug("Instantiating %s VNFRS for %s", len(vnfrs
), scale_instance
)
1694 scale_instance
.operational_status
= "vnf_init_phase"
1695 yield from self
.update_state()
1698 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
,
1699 group
, scale_instance
, vnfrs
)
1701 self
._log
.error("Pre scale out config for scale group {} ({}) failed".
1702 format(group
.name
, index
))
1703 scale_instance
.operational_status
= "failed"
1705 yield from self
.instantiate_vnfs(vnfrs
)
1707 except Exception as e
:
1708 self
._log
.exception("Failed to begin instantiatiation of vnfs for scale group {}: {}".
1709 format(group
.name
, e
))
1710 self
._log
.exception(e
)
1711 scale_instance
.operational_status
= "failed"
1713 yield from self
.update_state()
1715 yield from instantiate_instance()
1718 def delete_scale_group_instance(self
, group_name
, index
):
1719 group
= self
._scaling
_groups
[group_name
]
1720 scale_instance
= group
.get_instance(index
)
1721 if scale_instance
.is_default
:
1722 raise ScalingOperationError("Cannot terminate a default scaling group instance")
1724 scale_instance
.operational_status
= "terminate"
1725 yield from self
.update_state()
1728 def terminate_instance():
1729 self
._log
.debug("Terminating %s VNFRS" % scale_instance
)
1730 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_IN
,
1731 group
, scale_instance
)
1733 self
._log
.error("Pre scale in config for scale group {} ({}) failed".
1734 format(group
.name
, index
))
1736 # Going ahead with terminate, even if there is an error in pre-scale-in config
1737 # as this could be result of scale out failure and we need to cleanup this group
1738 yield from self
.terminate_vnfrs(scale_instance
.vnfrs
)
1739 group
.delete_instance(index
)
1741 scale_instance
.operational_status
= "vnf_terminate_phase"
1742 yield from self
.update_state()
1744 yield from terminate_instance()
1747 def _update_scale_group_instances_status(self
):
1749 def post_scale_out_task(group
, instance
):
1750 # Apply post scale out config once all VNFRs are active
1751 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_OUT
,
1753 instance
.operational_status
= "running"
1755 self
._log
.debug("Scale out for group {} and instance {} succeeded".
1756 format(group
.name
, instance
.instance_id
))
1758 self
._log
.error("Post scale out config for scale group {} ({}) failed".
1759 format(group
.name
, instance
.instance_id
))
1761 yield from self
.update_state()
1763 group_instances
= {group
: group
.instances
for group
in self
._scaling
_groups
.values()}
1764 for group
, instances
in group_instances
.items():
1765 self
._log
.debug("Updating %s instance status", group
)
1766 for instance
in instances
:
1767 instance_vnf_state_list
= [vnfr
.state
for vnfr
in instance
.vnfrs
]
1768 self
._log
.debug("Got vnfr instance states: %s", instance_vnf_state_list
)
1769 if instance
.operational_status
== "vnf_init_phase":
1770 if all([state
== VnfRecordState
.ACTIVE
for state
in instance_vnf_state_list
]):
1771 instance
.operational_status
= "running"
1773 # Create a task for post scale out to allow us to sleep before attempting
1774 # to configure newly created VM's
1775 self
._loop
.create_task(post_scale_out_task(group
, instance
))
1777 elif any([state
== VnfRecordState
.FAILED
for state
in instance_vnf_state_list
]):
1778 self
._log
.debug("Scale out for group {} and instance {} failed".
1779 format(group
.name
, instance
.instance_id
))
1780 instance
.operational_status
= "failed"
1782 elif instance
.operational_status
== "vnf_terminate_phase":
1783 if all([state
== VnfRecordState
.TERMINATED
for state
in instance_vnf_state_list
]):
1784 instance
.operational_status
= "terminated"
1785 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_IN
,
1788 self
._log
.debug("Scale in for group {} and instance {} succeeded".
1789 format(group
.name
, instance
.instance_id
))
1791 self
._log
.error("Post scale in config for scale group {} ({}) failed".
1792 format(group
.name
, instance
.instance_id
))
1794 def create_vnffgs(self
):
1795 """ This function creates VNFFGs for every VNFFG in the NSD
1796 associated with this NSR"""
1798 for vnffgd
in self
.nsd_msg
.vnffgd
:
1799 self
._log
.debug("Found vnffgd %s in nsr id %s", vnffgd
, self
.id)
1800 vnffgr
= VnffgRecord(self
._dts
,
1803 self
._nsm
._vnffgmgr
,
1807 self
._sdn
_account
_name
1809 self
._vnffgrs
[vnffgr
.id] = vnffgr
1811 def resolve_vld_ip_profile(self
, nsd_msg
, vld
):
1812 self
._log
.debug("Receieved ip profile ref is %s",vld
.ip_profile_ref
)
1813 if not vld
.has_field('ip_profile_ref'):
1815 profile
= [profile
for profile
in nsd_msg
.ip_profiles
if profile
.name
== vld
.ip_profile_ref
]
1816 return profile
[0] if profile
else None
1819 def _create_vls(self
, vld
, cloud_account
,om_datacenter
):
1820 """Create a VLR in the cloud account specified using the given VLD
1824 cloud_account : Cloud account name
1829 vlr
= yield from VirtualLinkRecord
.create_record(
1837 self
.resolve_vld_ip_profile(self
.nsd_msg
, vld
),
1839 restart_mode
=self
.restart_mode
)
1843 def _extract_cloud_accounts_for_vl(self
, vld
):
1845 Extracts the list of cloud accounts from the NS Config obj
1848 1. Cloud accounts based connection point (vnf_cloud_account_map)
1850 vld : VLD yang object
1855 cloud_account_list
= []
1857 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1858 # Handle case where cloud_account is None
1860 for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1861 if vnf
.cloud_account
is not None or vnf
.om_datacenter
is not None:
1862 vnf_cloud_map
[vnf
.member_vnf_index_ref
] = (vnf
.cloud_account
,vnf
.om_datacenter
)
1864 for vnfc
in vld
.vnfd_connection_point_ref
:
1865 cloud_account
= vnf_cloud_map
.get(
1866 vnfc
.member_vnf_index_ref
,
1867 (self
.cloud_account_name
,self
.om_datacenter_name
))
1869 cloud_account_list
.append(cloud_account
)
1871 if self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1872 for vld_map
in self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1873 if vld_map
.vld_id_ref
== vld
.id:
1874 for cloud_account
in vld_map
.cloud_accounts
:
1875 cloud_account_list
.extend((cloud_account
,None))
1876 for om_datacenter
in vld_map
.om_datacenters
:
1877 cloud_account_list
.extend((None,om_datacenter
))
1879 # If no config has been provided then fall-back to the default
1881 if not cloud_account_list
:
1882 cloud_account_list
= [(self
.cloud_account_name
,self
.om_datacenter_name
)]
1884 self
._log
.debug("VL {} cloud accounts: {}".
1885 format(vld
.name
, cloud_account_list
))
1886 return set(cloud_account_list
)
1889 def create_vls(self
):
1890 """ This function creates VLs for every VLD in the NSD
1891 associated with this NSR"""
1892 for vld
in self
.nsd_msg
.vld
:
1893 self
._log
.debug("Found vld %s in nsr id %s", vld
, self
.id)
1894 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1895 for cloud_account
,om_datacenter
in cloud_account_list
:
1896 vlr
= yield from self
._create
_vls
(vld
, cloud_account
,om_datacenter
)
1897 self
._vlrs
.append(vlr
)
1901 def create_vl_instance(self
, vld
):
1902 self
._log
.debug("Create VL for {}: {}".format(self
.id, vld
.as_dict()))
1903 # Check if the VL is already present
1905 for vl
in self
._vlrs
:
1906 if vl
.vld_msg
.id == vld
.id:
1907 self
._log
.debug("The VLD %s already in NSR %s as VLR %s with status %s",
1908 vld
.id, self
.id, vl
.id, vl
.state
)
1910 if vlr
.state
!= VlRecordState
.TERMINATED
:
1911 err_msg
= "VLR for VL %s in NSR %s already instantiated", \
1913 self
._log
.error(err_msg
)
1914 raise NsrVlUpdateError(err_msg
)
1918 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1919 for account
,om_datacenter
in cloud_account_list
:
1920 vlr
= yield from self
._create
_vls
(vld
, account
,om_datacenter
)
1921 self
._vlrs
.append(vlr
)
1923 vlr
.state
= VlRecordState
.INSTANTIATION_PENDING
1924 yield from self
.update_state()
1927 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1928 vlr
.state
= VlRecordState
.ACTIVE
1930 except Exception as e
:
1931 err_msg
= "Error instantiating VL for NSR {} and VLD {}: {}". \
1932 format(self
.id, vld
.id, e
)
1933 self
._log
.error(err_msg
)
1934 self
._log
.exception(e
)
1935 vlr
.state
= VlRecordState
.FAILED
1937 yield from self
.update_state()
1940 def delete_vl_instance(self
, vld
):
1941 for vlr
in self
._vlrs
:
1942 if vlr
.vld_msg
.id == vld
.id:
1943 self
._log
.debug("Found VLR %s for VLD %s in NSR %s",
1944 vlr
.id, vld
.id, self
.id)
1945 vlr
.state
= VlRecordState
.TERMINATE_PENDING
1946 yield from self
.update_state()
1949 yield from self
.nsm_plugin
.terminate_vl(vlr
)
1950 vlr
.state
= VlRecordState
.TERMINATED
1951 self
._vlrs
.remove(vlr
)
1953 except Exception as e
:
1954 err_msg
= "Error terminating VL for NSR {} and VLD {}: {}". \
1955 format(self
.id, vld
.id, e
)
1956 self
._log
.error(err_msg
)
1957 self
._log
.exception(e
)
1958 vlr
.state
= VlRecordState
.FAILED
1960 yield from self
.update_state()
1964 def create_vnfs(self
, config_xact
):
1966 This function creates VNFs for every VNF in the NSD
1967 associated with this NSR
1969 self
._log
.debug("Creating %u VNFs associated with this NS id %s",
1970 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
1972 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1973 if not const_vnfd
.start_by_default
:
1974 self
._log
.debug("start_by_default set to False in constituent VNF (%s). Skipping start.",
1975 const_vnfd
.member_vnf_index
)
1978 vnfd_msg
= self
._get
_vnfd
(const_vnfd
.vnfd_id_ref
, config_xact
)
1979 cloud_account_name
,om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd
.member_vnf_index
)
1980 if cloud_account_name
is None:
1981 cloud_account_name
= self
.cloud_account_name
1982 yield from self
.create_vnf_record(vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
)
1985 def get_placement_groups(self
, vnfd_msg
, const_vnfd
):
1986 placement_groups
= []
1987 for group
in self
.nsd_msg
.placement_groups
:
1988 for member_vnfd
in group
.member_vnfd
:
1989 if (member_vnfd
.vnfd_id_ref
== vnfd_msg
.id) and \
1990 (member_vnfd
.member_vnf_index_ref
== const_vnfd
.member_vnf_index
):
1991 group_info
= self
.resolve_placement_group_cloud_construct(group
)
1992 if group_info
is None:
1993 self
._log
.error("Could not resolve cloud-construct for placement group: %s", group
.name
)
1994 ### raise PlacementGroupError("Could not resolve cloud-construct for placement group: {}".format(group.name))
1996 self
._log
.info("Successfully resolved cloud construct for placement group: %s for VNF: %s (Member Index: %s)",
1999 const_vnfd
.member_vnf_index
)
2000 placement_groups
.append(group_info
)
2001 return placement_groups
2004 def create_vnf_record(self
, vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
, group_name
=None, group_instance_id
=None):
2005 # Fetch the VNFD associated with this VNF
2006 placement_groups
= self
.get_placement_groups(vnfd_msg
, const_vnfd
)
2007 self
._log
.info("Cloud Account for VNF %d is %s",const_vnfd
.member_vnf_index
,cloud_account_name
)
2008 self
._log
.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s",
2010 const_vnfd
.member_vnf_index
,
2011 [ group
.name
for group
in placement_groups
])
2012 vnfr
= yield from VirtualNetworkFunctionRecord
.create_record(self
._dts
,
2025 restart_mode
=self
.restart_mode
,
2027 if vnfr
.id in self
._vnfrs
:
2028 err
= "VNF with VNFR id %s already in vnf list" % (vnfr
.id,)
2029 raise NetworkServiceRecordError(err
)
2031 self
._vnfrs
[vnfr
.id] = vnfr
2032 self
._nsm
.vnfrs
[vnfr
.id] = vnfr
2034 yield from vnfr
.set_config_status(NsrYang
.ConfigStates
.INIT
)
2036 self
._log
.debug("Added VNFR %s to NSM VNFR list with id %s",
2042 def create_param_pools(self
):
2043 for param_pool
in self
.nsd_msg
.parameter_pool
:
2044 self
._log
.debug("Found parameter pool %s in nsr id %s", param_pool
, self
.id)
2046 start_value
= param_pool
.range.start_value
2047 end_value
= param_pool
.range.end_value
2048 if end_value
< start_value
:
2049 raise NetworkServiceRecordError(
2050 "Parameter pool %s has invalid range (start: {}, end: {})".format(
2051 start_value
, end_value
2055 self
._param
_pools
[param_pool
.name
] = config_value_pool
.ParameterValuePool(
2058 range(start_value
, end_value
)
2062 def fetch_vnfr(self
, vnfr_path
):
2063 """ Fetch VNFR record """
2065 self
._log
.debug("Fetching VNFR with key %s while instantiating %s",
2067 res_iter
= yield from self
._dts
.query_read(vnfr_path
, rwdts
.XactFlag
.MERGE
)
2069 for ent
in res_iter
:
2070 res
= yield from ent
2076 def instantiate_vnfs(self
, vnfrs
):
2078 This function instantiates VNFs for every VNF in this Network Service
2080 self
._log
.debug("Instantiating %u VNFs in NS %s", len(vnfrs
), self
.id)
2082 self
._log
.debug("Instantiating VNF: %s in NS %s", vnf
, self
.id)
2083 yield from self
.nsm_plugin
.instantiate_vnf(self
, vnf
)
2086 def instantiate_vnffgs(self
):
2088 This function instantiates VNFFGs for every VNFFG in this Network Service
2090 self
._log
.debug("Instantiating %u VNFFGs in NS %s",
2091 len(self
.nsd_msg
.vnffgd
), self
.id)
2092 for _
, vnfr
in self
.vnfrs
.items():
2093 while vnfr
.state
in [VnfRecordState
.INSTANTIATION_PENDING
, VnfRecordState
.INIT
]:
2094 self
._log
.debug("Received vnfr state for vnfr %s is %s; retrying",vnfr
.name
,vnfr
.state
)
2095 yield from asyncio
.sleep(2, loop
=self
._loop
)
2096 if vnfr
.state
== VnfRecordState
.ACTIVE
:
2097 self
._log
.debug("Received vnfr state for vnfr %s is %s ",vnfr
.name
,vnfr
.state
)
2100 self
._log
.debug("Received vnfr state for vnfr %s is %s; failing vnffg creation",vnfr
.name
,vnfr
.state
)
2101 self
._vnffgr
_state
= VnffgRecordState
.FAILED
2104 self
._log
.info("Waiting for 90 seconds for VMs to come up")
2105 yield from asyncio
.sleep(90, loop
=self
._loop
)
2106 self
._log
.info("Starting VNFFG orchestration")
2107 for vnffg
in self
._vnffgrs
.values():
2108 self
._log
.debug("Instantiating VNFFG: %s in NS %s", vnffg
, self
.id)
2109 yield from vnffg
.instantiate()
2112 def instantiate_scaling_instances(self
, config_xact
):
2113 """ Instantiate any default scaling instances in this Network Service """
2114 for group
in self
._scaling
_groups
.values():
2115 for i
in range(group
.min_instance_count
):
2116 self
._log
.debug("Instantiating %s default scaling instance %s", group
, i
)
2117 yield from self
.create_scale_group_instance(
2118 group
.name
, i
, config_xact
, is_default
=True
2121 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2122 if group_msg
.scaling_group_name_ref
!= group
.name
:
2125 for instance
in group_msg
.instance
:
2126 self
._log
.debug("Reloading %s scaling instance %s", group_msg
, instance
.id)
2127 yield from self
.create_scale_group_instance(
2128 group
.name
, instance
.id, config_xact
, is_default
=False
2131 def has_scaling_instances(self
):
2132 """ Return boolean indicating if the network service has default scaling groups """
2133 for group
in self
._scaling
_groups
.values():
2134 if group
.min_instance_count
> 0:
2137 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2138 if len(group_msg
.instance
) > 0:
2145 """ This function publishes this NSR """
2146 self
._nsr
_msg
= self
.create_msg()
2148 self
._log
.debug("Publishing the NSR with xpath %s and nsr %s",
2152 if self
._debug
_running
:
2153 self
._log
.debug("Publishing NSR in RUNNING state!")
2156 with self
._dts
.transaction() as xact
:
2157 yield from self
._nsm
.nsr_handler
.update(xact
, self
.nsr_xpath
, self
._nsr
_msg
)
2158 if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
:
2159 self
._debug
_running
= True
2162 def unpublish(self
, xact
):
2163 """ Unpublish this NSR object """
2164 self
._log
.debug("Unpublishing Network service id %s", self
.id)
2165 yield from self
._nsm
.nsr_handler
.delete(xact
, self
.nsr_xpath
)
2168 def nsr_xpath(self
):
2169 """ Returns the xpath associated with this NSR """
2171 "D,/nsr:ns-instance-opdata" +
2172 "/nsr:nsr[nsr:ns-instance-config-ref = '{}']"
2176 def xpath_from_nsr(nsr
):
2177 """ Returns the xpath associated with this NSR op data"""
2178 return (NetworkServiceRecord
.XPATH
+
2179 "[nsr:ns-instance-config-ref = '{}']").format(nsr
.id)
2182 def nsd_xpath(self
):
2183 """ Return NSD config xpath."""
2185 "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}']"
2186 ).format(self
.nsd_id
)
2189 def instantiate(self
, config_xact
):
2190 """"Instantiates a NetworkServiceRecord.
2192 This function instantiates a Network service
2193 which involves the following steps,
2195 * Instantiate every VL in NSD by sending create VLR request to DTS.
2196 * Instantiate every VNF in NSD by sending create VNF reuqest to DTS.
2197 * Publish the NSR details to DTS
2200 nsr: The NSR configuration request containing nsr-id and nsd
2201 config_xact: The configuration transaction which initiated the instatiation
2204 NetworkServiceRecordError if the NSR creation fails
2210 self
._log
.debug("Instantiating NS - %s xact - %s", self
, config_xact
)
2212 # Move the state to INIITALIZING
2213 self
.set_state(NetworkServiceRecordState
.INIT
)
2215 event_descr
= "Instantiation Request Received NSR Id:%s" % self
.id
2216 self
.record_event("instantiating", event_descr
)
2219 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
2222 # Update ref count if nsd present in catalog
2223 self
._nsm
.get_nsd_ref(self
.nsd_id
)
2225 except NetworkServiceDescriptorError
:
2226 # This could be an NSD not in the nsd-catalog
2229 # Merge any config and initial config primitive values
2230 self
.config_store
.merge_nsd_config(self
.nsd_msg
)
2231 self
._log
.debug("Merged NSD: {}".format(self
.nsd_msg
.as_dict()))
2233 event_descr
= "Fetched NSD with descriptor id %s" % self
.nsd_id
2234 self
.record_event("nsd-fetched", event_descr
)
2236 if self
._nsd
is None:
2237 msg
= "Failed to fetch NSD with nsd-id [%s] for nsr-id %s"
2238 self
._log
.debug(msg
, self
.nsd_id
, self
.id)
2239 raise NetworkServiceRecordError(self
)
2241 self
._log
.debug("Got nsd result %s", self
._nsd
)
2243 # Substitute any input parameters
2244 self
.substitute_input_parameters(self
._nsd
, self
._nsr
_cfg
_msg
)
2247 yield from self
.create(config_xact
)
2249 # Publish the NSR to DTS
2250 yield from self
.publish()
2253 def do_instantiate():
2255 Instantiate network service
2257 self
._log
.debug("Instantiating VLs nsr id [%s] nsd id [%s]",
2258 self
.id, self
.nsd_id
)
2260 # instantiate the VLs
2261 event_descr
= ("Instantiating %s external VLs for NSR id %s" %
2262 (len(self
.nsd_msg
.vld
), self
.id))
2263 self
.record_event("begin-external-vls-instantiation", event_descr
)
2265 self
.set_state(NetworkServiceRecordState
.VL_INIT_PHASE
)
2267 yield from self
.instantiate_vls()
2269 # Publish the NSR to DTS
2270 yield from self
.publish()
2272 event_descr
= ("Finished instantiating %s external VLs for NSR id %s" %
2273 (len(self
.nsd_msg
.vld
), self
.id))
2274 self
.record_event("end-external-vls-instantiation", event_descr
)
2276 self
.set_state(NetworkServiceRecordState
.VNF_INIT_PHASE
)
2278 self
._log
.debug("Instantiating VNFs ...... nsr[%s], nsd[%s]",
2279 self
.id, self
.nsd_id
)
2281 # instantiate the VNFs
2282 event_descr
= ("Instantiating %s VNFS for NSR id %s" %
2283 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2285 self
.record_event("begin-vnf-instantiation", event_descr
)
2287 yield from self
.instantiate_vnfs(self
._vnfrs
.values())
2289 self
._log
.debug(" Finished instantiating %d VNFs for NSR id %s",
2290 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
2292 event_descr
= ("Finished instantiating %s VNFs for NSR id %s" %
2293 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2294 self
.record_event("end-vnf-instantiation", event_descr
)
2296 if len(self
.vnffgrs
) > 0:
2297 #self.set_state(NetworkServiceRecordState.VNFFG_INIT_PHASE)
2298 event_descr
= ("Instantiating %s VNFFGS for NSR id %s" %
2299 (len(self
.nsd_msg
.vnffgd
), self
.id))
2301 self
.record_event("begin-vnffg-instantiation", event_descr
)
2303 yield from self
.instantiate_vnffgs()
2305 event_descr
= ("Finished instantiating %s VNFFGDs for NSR id %s" %
2306 (len(self
.nsd_msg
.vnffgd
), self
.id))
2307 self
.record_event("end-vnffg-instantiation", event_descr
)
2309 if self
.has_scaling_instances():
2310 event_descr
= ("Instantiating %s Scaling Groups for NSR id %s" %
2311 (len(self
._scaling
_groups
), self
.id))
2313 self
.record_event("begin-scaling-group-instantiation", event_descr
)
2314 yield from self
.instantiate_scaling_instances(config_xact
)
2315 self
.record_event("end-scaling-group-instantiation", event_descr
)
2317 # Give the plugin a chance to deploy the network service now that all
2318 # virtual links and vnfs are instantiated
2319 yield from self
.nsm_plugin
.deploy(self
._nsr
_msg
)
2321 self
._log
.debug("Publishing NSR...... nsr[%s], nsd[%s]",
2322 self
.id, self
.nsd_id
)
2324 # Publish the NSR to DTS
2325 yield from self
.publish()
2327 self
._log
.debug("Published NSR...... nsr[%s], nsd[%s]",
2328 self
.id, self
.nsd_id
)
2330 def on_instantiate_done(fut
):
2331 # If the do_instantiate fails, then publish NSR with failed result
2332 if fut
.exception() is not None:
2333 self
._log
.error("NSR instantiation failed for NSR id %s: %s", self
.id, str(fut
.exception()))
2334 self
._loop
.create_task(self
.instantiation_failed(failed_reason
=str(fut
.exception())))
2336 instantiate_task
= self
._loop
.create_task(do_instantiate())
2337 instantiate_task
.add_done_callback(on_instantiate_done
)
2340 def set_config_status(self
, status
, status_details
=None):
2341 if self
.config_status
!= status
:
2342 self
._log
.debug("Updating NSR {} status for {} to {}".
2343 format(self
.name
, self
.config_status
, status
))
2344 self
._config
_status
= status
2345 self
._config
_status
_details
= status_details
2347 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2348 self
.record_event("config-failed", "NS configuration failed",
2349 evt_details
=self
._config
_status
_details
)
2351 yield from self
.publish()
2354 def is_active(self
):
2355 """ This NS is active """
2356 self
.set_state(NetworkServiceRecordState
.RUNNING
)
2360 # Publish the NSR to DTS
2361 self
._log
.debug("Network service %s is active ", self
.id)
2362 self
._is
_active
= True
2364 event_descr
= "NSR in running state for NSR id %s" % self
.id
2365 self
.record_event("ns-running", event_descr
)
2367 yield from self
.publish()
2370 def instantiation_failed(self
, failed_reason
=None):
2371 """ The NS instantiation failed"""
2372 self
._log
.error("Network service id:%s, name:%s instantiation failed",
2374 self
.set_state(NetworkServiceRecordState
.FAILED
)
2376 event_descr
= "Instantiation of NS %s failed" % self
.id
2377 self
.record_event("ns-failed", event_descr
, evt_details
=failed_reason
)
2379 # Publish the NSR to DTS
2380 yield from self
.publish()
2383 def terminate_vnfrs(self
, vnfrs
):
2384 """ Terminate VNFRS in this network service """
2385 self
._log
.debug("Terminating VNFs in network service %s", self
.id)
2387 yield from self
.nsm_plugin
.terminate_vnf(vnfr
)
2390 def terminate(self
):
2391 """ Terminate a NetworkServiceRecord."""
2392 def terminate_vnffgrs():
2393 """ Terminate VNFFGRS in this network service """
2394 self
._log
.debug("Terminating VNFFGRs in network service %s", self
.id)
2395 for vnffgr
in self
.vnffgrs
.values():
2396 yield from vnffgr
.terminate()
2398 def terminate_vlrs():
2399 """ Terminate VLRs in this netork service """
2400 self
._log
.debug("Terminating VLs in network service %s", self
.id)
2401 for vlr
in self
.vlrs
:
2402 yield from self
.nsm_plugin
.terminate_vl(vlr
)
2403 vlr
.state
= VlRecordState
.TERMINATED
2405 self
._log
.debug("Terminating network service id %s", self
.id)
2407 # Move the state to TERMINATE
2408 self
.set_state(NetworkServiceRecordState
.TERMINATE
)
2409 event_descr
= "Terminate being processed for NS Id:%s" % self
.id
2410 self
.record_event("terminate", event_descr
)
2412 # Move the state to VNF_TERMINATE_PHASE
2413 self
._log
.debug("Terminating VNFFGs in NS ID: %s", self
.id)
2414 self
.set_state(NetworkServiceRecordState
.VNFFG_TERMINATE_PHASE
)
2415 event_descr
= "Terminating VNFFGS in NS Id:%s" % self
.id
2416 self
.record_event("terminating-vnffgss", event_descr
)
2417 yield from terminate_vnffgrs()
2419 # Move the state to VNF_TERMINATE_PHASE
2420 self
.set_state(NetworkServiceRecordState
.VNF_TERMINATE_PHASE
)
2421 event_descr
= "Terminating VNFS in NS Id:%s" % self
.id
2422 self
.record_event("terminating-vnfs", event_descr
)
2423 yield from self
.terminate_vnfrs(self
.vnfrs
.values())
2425 # Move the state to VL_TERMINATE_PHASE
2426 self
.set_state(NetworkServiceRecordState
.VL_TERMINATE_PHASE
)
2427 event_descr
= "Terminating VLs in NS Id:%s" % self
.id
2428 self
.record_event("terminating-vls", event_descr
)
2429 yield from terminate_vlrs()
2431 yield from self
.nsm_plugin
.terminate_ns(self
)
2433 # Move the state to TERMINATED
2434 self
.set_state(NetworkServiceRecordState
.TERMINATED
)
2435 event_descr
= "Terminated NS Id:%s" % self
.id
2436 self
.record_event("terminated", event_descr
)
2439 """"Enable a NetworkServiceRecord."""
2443 """"Disable a NetworkServiceRecord."""
2446 def map_config_status(self
):
2447 self
._log
.debug("Config status for ns {} is {}".
2448 format(self
.name
, self
._config
_status
))
2449 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURING
:
2450 return 'configuring'
2451 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2455 def vl_phase_completed(self
):
2456 """ Are VLs created in this NS?"""
2457 return self
._vl
_phase
_completed
2459 def vnf_phase_completed(self
):
2460 """ Are VLs created in this NS?"""
2461 return self
._vnf
_phase
_completed
2463 def create_msg(self
):
2464 """ The network serice record as a message """
2465 nsr_dict
= {"ns_instance_config_ref": self
.id}
2466 nsr
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr
.from_dict(nsr_dict
)
2467 #nsr.cloud_account = self.cloud_account_name
2468 nsr
.sdn_account
= self
._sdn
_account
_name
2469 nsr
.name_ref
= self
.name
2470 nsr
.nsd_ref
= self
.nsd_id
2471 nsr
.nsd_name_ref
= self
.nsd_msg
.name
2472 nsr
.operational_events
= self
._op
_status
.msg
2473 nsr
.operational_status
= self
._op
_status
.yang_str()
2474 nsr
.config_status
= self
.map_config_status()
2475 nsr
.config_status_details
= self
._config
_status
_details
2476 nsr
.create_time
= self
._create
_time
2478 for cfg_prim
in self
.nsd_msg
.service_primitive
:
2479 cfg_prim
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_ServicePrimitive
.from_dict(
2481 nsr
.service_primitive
.append(cfg_prim
)
2483 for init_cfg
in self
.nsd_msg
.initial_config_primitive
:
2484 prim
= NsrYang
.NsrInitialConfigPrimitive
.from_dict(
2486 nsr
.initial_config_primitive
.append(prim
)
2488 if self
.vl_phase_completed():
2489 for vlr
in self
.vlrs
:
2490 nsr
.vlr
.append(vlr
.create_nsr_vlr_msg(self
.vnfrs
.values()))
2492 if self
.vnf_phase_completed():
2493 for vnfr_id
in self
.vnfrs
:
2494 nsr
.constituent_vnfr_ref
.append(self
.vnfrs
[vnfr_id
].const_vnfr_msg
)
2495 for vnffgr
in self
.vnffgrs
.values():
2496 nsr
.vnffgr
.append(vnffgr
.fetch_vnffgr())
2497 for scaling_group
in self
._scaling
_groups
.values():
2498 nsr
.scaling_group_record
.append(scaling_group
.create_record_msg())
2502 def all_vnfs_active(self
):
2503 """ Are all VNFS in this NS active? """
2504 for _
, vnfr
in self
.vnfrs
.items():
2505 if vnfr
.active
is not True:
2510 def update_state(self
):
2511 """ Re-evaluate this NS's state """
2512 curr_state
= self
._op
_status
.state
2514 if curr_state
== NetworkServiceRecordState
.TERMINATED
:
2515 self
._log
.debug("NS (%s) in terminated state, not updating state", self
.id)
2518 new_state
= NetworkServiceRecordState
.RUNNING
2519 self
._log
.info("Received update_state for nsr: %s, curr-state: %s",
2520 self
.id, curr_state
)
2522 # Check all the VNFRs are present
2523 for _
, vnfr
in self
.vnfrs
.items():
2524 if vnfr
.state
in [VnfRecordState
.ACTIVE
, VnfRecordState
.TERMINATED
]:
2526 elif vnfr
.state
== VnfRecordState
.FAILED
:
2527 if vnfr
._prev
_state
!= vnfr
.state
:
2528 event_descr
= "Instantiation of VNF %s failed" % vnfr
.id
2529 event_error_details
= vnfr
.state_failed_reason
2530 self
.record_event("vnf-failed", event_descr
, evt_details
=event_error_details
)
2531 vnfr
.set_state(VnfRecordState
.FAILED
)
2533 self
._log
.info("VNF state did not change, curr=%s, prev=%s",
2534 vnfr
.state
, vnfr
._prev
_state
)
2535 new_state
= NetworkServiceRecordState
.FAILED
2538 self
._log
.info("VNF %s in NSR %s is still not active; current state is: %s",
2539 vnfr
.id, self
.id, vnfr
.state
)
2540 new_state
= curr_state
2542 # If new state is RUNNING; check all VLs
2543 if new_state
== NetworkServiceRecordState
.RUNNING
:
2544 for vl
in self
.vlrs
:
2546 if vl
.state
in [VlRecordState
.ACTIVE
, VlRecordState
.TERMINATED
]:
2548 elif vl
.state
== VlRecordState
.FAILED
:
2549 if vl
.prev_state
!= vl
.state
:
2550 event_descr
= "Instantiation of VL %s failed" % vl
.id
2551 event_error_details
= vl
.state_failed_reason
2552 self
.record_event("vl-failed", event_descr
, evt_details
=event_error_details
)
2553 vl
.prev_state
= vl
.state
2555 self
._log
.debug("VL %s already in failed state")
2557 if vl
.state
in [VlRecordState
.INSTANTIATION_PENDING
, VlRecordState
.INIT
]:
2558 new_state
= NetworkServiceRecordState
.VL_INSTANTIATE
2561 if vl
.state
in [VlRecordState
.TERMINATE_PENDING
]:
2562 new_state
= NetworkServiceRecordState
.VL_TERMINATE
2565 # If new state is RUNNING; check VNFFGRs are also active
2566 if new_state
== NetworkServiceRecordState
.RUNNING
:
2567 for _
, vnffgr
in self
.vnffgrs
.items():
2568 self
._log
.info("Checking vnffgr state for nsr %s is: %s",
2569 self
.id, vnffgr
.state
)
2570 if vnffgr
.state
== VnffgRecordState
.ACTIVE
:
2572 elif vnffgr
.state
== VnffgRecordState
.FAILED
:
2573 event_descr
= "Instantiation of VNFFGR %s failed" % vnffgr
.id
2574 self
.record_event("vnffg-failed", event_descr
)
2575 new_state
= NetworkServiceRecordState
.FAILED
2578 self
._log
.info("VNFFGR %s in NSR %s is still not active; current state is: %s",
2579 vnffgr
.id, self
.id, vnffgr
.state
)
2580 new_state
= curr_state
2582 # Update all the scaling group instance operational status to
2583 # reflect the state of all VNFR within that instance
2584 yield from self
._update
_scale
_group
_instances
_status
()
2586 for _
, group
in self
._scaling
_groups
.items():
2587 if group
.state
== scale_group
.ScaleGroupState
.SCALING_OUT
:
2588 new_state
= NetworkServiceRecordState
.SCALING_OUT
2590 elif group
.state
== scale_group
.ScaleGroupState
.SCALING_IN
:
2591 new_state
= NetworkServiceRecordState
.SCALING_IN
2594 if new_state
!= curr_state
:
2595 self
._log
.debug("Changing state of Network service %s from %s to %s",
2596 self
.id, curr_state
, new_state
)
2597 if new_state
== NetworkServiceRecordState
.RUNNING
:
2598 yield from self
.is_active()
2599 elif new_state
== NetworkServiceRecordState
.FAILED
:
2600 # If the NS is already active and we entered scaling_in, scaling_out,
2601 # do not mark the NS as failing if scaling operation failed.
2602 if curr_state
in [NetworkServiceRecordState
.SCALING_OUT
,
2603 NetworkServiceRecordState
.SCALING_IN
] and self
._is
_active
:
2604 new_state
= NetworkServiceRecordState
.RUNNING
2605 self
.set_state(new_state
)
2607 yield from self
.instantiation_failed()
2609 self
.set_state(new_state
)
2611 yield from self
.publish()
2614 class InputParameterSubstitution(object):
2616 This class is responsible for substituting input parameters into an NSD.
2619 def __init__(self
, log
):
2620 """Create an instance of InputParameterSubstitution
2623 log - a logger for this object to use
2628 def __call__(self
, nsd
, nsr_config
):
2629 """Substitutes input parameters from the NSR config into the NSD
2631 This call modifies the provided NSD with the input parameters that are
2632 contained in the NSR config.
2635 nsd - a GI NSD object
2636 nsr_config - a GI NSR config object
2639 if nsd
is None or nsr_config
is None:
2642 # Create a lookup of the xpath elements that this descriptor allows
2644 optional_input_parameters
= set()
2645 for input_parameter
in nsd
.input_parameter_xpath
:
2646 optional_input_parameters
.add(input_parameter
.xpath
)
2648 # Apply the input parameters to the descriptor
2649 if nsr_config
.input_parameter
:
2650 for param
in nsr_config
.input_parameter
:
2651 if param
.xpath
not in optional_input_parameters
:
2652 msg
= "tried to set an invalid input parameter ({})"
2653 self
.log
.error(msg
.format(param
.xpath
))
2657 "input-parameter:{} = {}".format(
2664 xpath
.setxattr(nsd
, param
.xpath
, param
.value
)
2666 except Exception as e
:
2667 self
.log
.exception(e
)
2670 class NetworkServiceDescriptor(object):
2672 Network service descriptor class
2675 def __init__(self
, dts
, log
, loop
, nsd
, nsm
):
2687 """ Returns nsd id """
2692 """ Returns name of nsd """
2693 return self
._nsd
.name
2696 def ref_count(self
):
2697 """ Returns reference count"""
2698 return self
._ref
_count
2701 """ Returns whether nsd is in use or not """
2702 return True if self
.ref_count
> 0 else False
2705 """ Take a reference on this object """
2706 self
._ref
_count
+= 1
2709 """ Release reference on this object """
2710 if self
.ref_count
< 1:
2711 msg
= ("Unref on a NSD object - nsd id %s, ref_count = %s" %
2712 (self
.id, self
.ref_count
))
2713 self
._log
.critical(msg
)
2714 raise NetworkServiceDescriptorError(msg
)
2715 self
._ref
_count
-= 1
2719 """ Return the message associated with this NetworkServiceDescriptor"""
2723 def path_for_id(nsd_id
):
2724 """ Return path for the passed nsd_id"""
2725 return "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}'".format(nsd_id
)
2728 """ Return the message associated with this NetworkServiceDescriptor"""
2729 return NetworkServiceDescriptor
.path_for_id(self
.id)
2731 def update(self
, nsd
):
2732 """ Update the NSD descriptor """
2736 class NsdDtsHandler(object):
2737 """ The network service descriptor DTS handler """
2738 XPATH
= "C,/nsd:nsd-catalog/nsd:nsd"
2740 def __init__(self
, dts
, log
, loop
, nsm
):
2750 """ Return registration handle """
2755 """ Register for Nsd create/update/delete/read requests from dts """
2757 def on_apply(dts
, acg
, xact
, action
, scratch
):
2758 """Apply the configuration"""
2759 is_recovery
= xact
.xact
is None and action
== rwdts
.AppconfAction
.INSTALL
2760 self
._log
.debug("Got nsd apply cfg (xact:%s) (action:%s)",
2762 # Create/Update an NSD record
2763 for cfg
in self
._regh
.get_xact_elements(xact
):
2764 # Only interested in those NSD cfgs whose ID was received in prepare callback
2765 if cfg
.id in scratch
.get('nsds', []) or is_recovery
:
2766 self
._nsm
.update_nsd(cfg
)
2768 scratch
.pop('nsds', None)
2770 return RwTypes
.RwStatus
.SUCCESS
2773 def delete_nsd_libs(nsd_id
):
2774 """ Remove any files uploaded with NSD and stored under $RIFT_ARTIFACTS/libs/<id> """
2776 rift_artifacts_dir
= os
.environ
['RIFT_ARTIFACTS']
2777 nsd_dir
= os
.path
.join(rift_artifacts_dir
, 'launchpad/libs', nsd_id
)
2779 if os
.path
.exists (nsd_dir
):
2780 shutil
.rmtree(nsd_dir
, ignore_errors
=True)
2781 except Exception as e
:
2782 self
._log
.error("Exception in cleaning up NSD libs {}: {}".
2784 self
._log
.excpetion(e
)
2787 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2788 """ Prepare callback from DTS for NSD config """
2790 self
._log
.info("Got nsd prepare - config received nsd id %s, msg %s",
2793 fref
= ProtobufC
.FieldReference
.alloc()
2794 fref
.goto_whole_message(msg
.to_pbcm())
2796 if fref
.is_field_deleted():
2797 # Delete an NSD record
2798 self
._log
.debug("Deleting NSD with id %s", msg
.id)
2799 if self
._nsm
.nsd_in_use(msg
.id):
2800 self
._log
.debug("Cannot delete NSD in use - %s", msg
.id)
2801 err
= "Cannot delete an NSD in use - %s" % msg
.id
2802 raise NetworkServiceDescriptorRefCountExists(err
)
2804 yield from delete_nsd_libs(msg
.id)
2805 self
._nsm
.delete_nsd(msg
.id)
2807 # Add this NSD to scratch to create/update in apply callback
2808 nsds
= scratch
.setdefault('nsds', [])
2810 # acg._scratch['nsds'].append(msg.id)
2812 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2815 "Registering for NSD config using xpath: %s",
2816 NsdDtsHandler
.XPATH
,
2819 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2820 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2821 # Need a list in scratch to store NSDs to create/update later
2822 # acg._scratch['nsds'] = list()
2823 self
._regh
= acg
.register(
2824 xpath
=NsdDtsHandler
.XPATH
,
2825 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
2826 on_prepare
=on_prepare
)
2829 class VnfdDtsHandler(object):
2830 """ DTS handler for VNFD config changes """
2831 XPATH
= "C,/vnfd:vnfd-catalog/vnfd:vnfd"
2833 def __init__(self
, dts
, log
, loop
, nsm
):
2842 """ DTS registration handle """
2847 """ Register for VNFD configuration"""
2850 def on_apply(dts
, acg
, xact
, action
, scratch
):
2851 """Apply the configuration"""
2852 self
._log
.debug("Got NSM VNFD apply (xact: %s) (action: %s)(scr: %s)",
2853 xact
, action
, scratch
)
2855 # Create/Update a VNFD record
2856 for cfg
in self
._regh
.get_xact_elements(xact
):
2857 # Only interested in those VNFD cfgs whose ID was received in prepare callback
2858 if cfg
.id in scratch
.get('vnfds', []):
2859 self
._nsm
.update_vnfd(cfg
)
2861 for cfg
in self
._regh
.elements
:
2862 if cfg
.id in scratch
.get('deleted_vnfds', []):
2863 yield from self
._nsm
.delete_vnfd(cfg
.id)
2865 scratch
.pop('vnfds', None)
2866 scratch
.pop('deleted_vnfds', None)
2869 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2870 """ on prepare callback """
2871 self
._log
.debug("Got on prepare for VNFD (path: %s) (action: %s) (msg: %s)",
2872 ks_path
.to_xpath(RwNsmYang
.get_schema()), xact_info
.query_action
, msg
)
2874 fref
= ProtobufC
.FieldReference
.alloc()
2875 fref
.goto_whole_message(msg
.to_pbcm())
2877 # Handle deletes in prepare_callback, but adds/updates in apply_callback
2878 if fref
.is_field_deleted():
2879 self
._log
.debug("Adding msg to deleted field")
2880 deleted_vnfds
= scratch
.setdefault('deleted_vnfds', [])
2881 deleted_vnfds
.append(msg
.id)
2883 # Add this VNFD to scratch to create/update in apply callback
2884 vnfds
= scratch
.setdefault('vnfds', [])
2885 vnfds
.append(msg
.id)
2887 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2890 "Registering for VNFD config using xpath: %s",
2891 VnfdDtsHandler
.XPATH
,
2893 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2894 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2895 # Need a list in scratch to store VNFDs to create/update later
2896 # acg._scratch['vnfds'] = list()
2897 # acg._scratch['deleted_vnfds'] = list()
2898 self
._regh
= acg
.register(
2899 xpath
=VnfdDtsHandler
.XPATH
,
2900 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY
,
2901 on_prepare
=on_prepare
)
2903 class NsrRpcDtsHandler(object):
2904 """ The network service instantiation RPC DTS handler """
2905 EXEC_NSR_CONF_XPATH
= "I,/nsr:start-network-service"
2906 EXEC_NSR_CONF_O_XPATH
= "O,/nsr:start-network-service"
2907 NETCONF_IP_ADDRESS
= "127.0.0.1"
2909 RESTCONF_PORT
= 8888
2910 NETCONF_USER
= "admin"
2911 NETCONF_PW
= "admin"
2912 REST_BASE_V2_URL
= 'https://{}:{}/v2/api/'.format("127.0.0.1",8888)
2914 def __init__(self
, dts
, log
, loop
, nsm
):
2921 self
._ns
_regh
= None
2923 self
._manager
= None
2924 self
._nsr
_config
_url
= NsrRpcDtsHandler
.REST_BASE_V2_URL
+ 'config/ns-instance-config'
2926 self
._model
= RwYang
.Model
.create_libncx()
2927 self
._model
.load_schema_ypbc(RwNsrYang
.get_schema())
2931 """ Return the NS manager instance """
2935 def wrap_netconf_config_xml(xml
):
2936 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
2940 def _connect(self
, timeout_secs
=240):
2942 start_time
= time
.time()
2943 while (time
.time() - start_time
) < timeout_secs
:
2946 self
._log
.debug("Attemping NsmTasklet netconf connection.")
2948 manager
= yield from ncclient
.asyncio_manager
.asyncio_connect(
2950 host
=NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
,
2951 port
=NsrRpcDtsHandler
.NETCONF_PORT
,
2952 username
=NsrRpcDtsHandler
.NETCONF_USER
,
2953 password
=NsrRpcDtsHandler
.NETCONF_PW
,
2955 look_for_keys
=False,
2956 hostkey_verify
=False,
2961 except ncclient
.transport
.errors
.SSHError
as e
:
2962 self
._log
.warning("Netconf connection to launchpad %s failed: %s",
2963 NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
, str(e
))
2965 yield from asyncio
.sleep(5, loop
=self
._loop
)
2967 raise NsrInstantiationFailed("Failed to connect to Launchpad within %s seconds" %
2970 def _apply_ns_instance_config(self
,payload_dict
):
2971 #self._log.debug("At apply NS instance config with payload %s",payload_dict)
2972 req_hdr
= {'accept':'application/vnd.yang.data+json','content-type':'application/vnd.yang.data+json'}
2973 response
=requests
.post(self
._nsr
_config
_url
, headers
=req_hdr
, auth
=('admin', 'admin'),data
=payload_dict
,verify
=False)
2978 """ Register for NS monitoring read from dts """
2980 def on_ns_config_prepare(xact_info
, action
, ks_path
, msg
):
2981 """ prepare callback from dts start-network-service"""
2982 assert action
== rwdts
.QueryAction
.RPC
2984 rpc_op
= NsrYang
.YangOutput_Nsr_StartNetworkService
.from_dict({
2985 "nsr_id":str(uuid
.uuid4())
2988 if not ('name' in rpc_ip
and 'nsd_ref' in rpc_ip
and ('cloud_account' in rpc_ip
or 'om_datacenter' in rpc_ip
)):
2989 self
._log
.error("Mandatory parameters name or nsd_ref or cloud account not found in start-network-service {}".format(rpc_ip
))
2992 self
._log
.debug("start-network-service RPC input: {}".format(rpc_ip
))
2995 # Add used value to the pool
2996 self
._log
.debug("RPC output: {}".format(rpc_op
))
2998 nsd_copy
= self
.nsm
.get_nsd(rpc_ip
.nsd_ref
)
3000 #if not self._manager:
3001 # self._manager = yield from self._connect()
3003 self
._log
.debug("Configuring ns-instance-config with name %s nsd-ref: %s",
3004 rpc_ip
.name
, rpc_ip
.nsd_ref
)
3006 ns_instance_config_dict
= {"id":rpc_op
.nsr_id
, "admin_status":"ENABLED"}
3007 ns_instance_config_copy_dict
= {k
:v
for k
, v
in rpc_ip
.as_dict().items()
3008 if k
in RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr().fields
}
3009 ns_instance_config_dict
.update(ns_instance_config_copy_dict
)
3011 ns_instance_config
= RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.from_dict(ns_instance_config_dict
)
3012 ns_instance_config
.nsd
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_Nsd()
3013 ns_instance_config
.nsd
.from_dict(nsd_copy
.msg
.as_dict())
3015 payload_dict
= ns_instance_config
.to_json(self
._model
)
3016 #xml = ns_instance_config.to_xml_v2(self._model)
3017 #netconf_xml = self.wrap_netconf_config_xml(xml)
3019 #self._log.debug("Sending configure ns-instance-config xml to %s: %s",
3020 # netconf_xml, NsrRpcDtsHandler.NETCONF_IP_ADDRESS)
3021 self
._log
.debug("Sending configure ns-instance-config json to %s: %s",
3022 self
._nsr
_config
_url
,ns_instance_config
)
3024 #response = yield from self._manager.edit_config(
3026 # config=netconf_xml,
3028 response
= yield from self
._loop
.run_in_executor(
3030 self
._apply
_ns
_instance
_config
,
3033 response
.raise_for_status()
3034 self
._log
.debug("Received edit config response: %s", response
.json())
3036 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
,
3037 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
,
3039 except Exception as e
:
3040 self
._log
.error("Exception processing the "
3041 "start-network-service: {}".format(e
))
3042 self
._log
.exception(e
)
3043 xact_info
.respond_xpath(rwdts
.XactRspCode
.NACK
,
3044 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
)
3047 hdl_ns
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_ns_config_prepare
,)
3049 with self
._dts
.group_create() as group
:
3050 self
._ns
_regh
= group
.register(xpath
=NsrRpcDtsHandler
.EXEC_NSR_CONF_XPATH
,
3052 flags
=rwdts
.Flag
.PUBLISHER
,
3056 class NsrDtsHandler(object):
3057 """ The network service DTS handler """
3058 NSR_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr"
3059 SCALE_INSTANCE_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr/nsr:scaling-group/nsr:instance"
3060 KEY_PAIR_XPATH
= "C,/nsr:key-pair"
3062 def __init__(self
, dts
, log
, loop
, nsm
):
3068 self
._nsr
_regh
= None
3069 self
._scale
_regh
= None
3070 self
._key
_pair
_regh
= None
3074 """ Return the NS manager instance """
3079 """ Register for Nsr create/update/delete/read requests from dts """
3081 def nsr_id_from_keyspec(ks
):
3082 nsr_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.schema().keyspec_to_entry(ks
)
3083 nsr_id
= nsr_path_entry
.key00
.id
3086 def group_name_from_keyspec(ks
):
3087 group_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
.schema().keyspec_to_entry(ks
)
3088 group_name
= group_path_entry
.key00
.scaling_group_name_ref
3091 def is_instance_in_reg_elements(nsr_id
, group_name
, instance_id
):
3092 """ Return boolean indicating if scaling group instance was already commited previously.
3094 By looking at the existing elements in this registration handle (elements not part
3095 of this current xact), we can tell if the instance was configured previously without
3096 keeping any application state.
3098 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3099 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3100 elem_group_name
= group_name_from_keyspec(keyspec
)
3102 if elem_nsr_id
!= nsr_id
or group_name
!= elem_group_name
:
3105 if instance_cfg
.id == instance_id
:
3110 def get_scale_group_instance_delta(nsr_id
, group_name
, xact
):
3111 delta
= {"added": [], "deleted": []}
3112 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3113 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3114 if elem_nsr_id
!= nsr_id
:
3117 elem_group_name
= group_name_from_keyspec(keyspec
)
3118 if elem_group_name
!= group_name
:
3121 delta
["added"].append(instance_cfg
.id)
3123 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(include_keyspec
=True):
3124 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3125 if elem_nsr_id
!= nsr_id
:
3128 elem_group_name
= group_name_from_keyspec(keyspec
)
3129 if elem_group_name
!= group_name
:
3132 if instance_cfg
.id in delta
["added"]:
3133 delta
["added"].remove(instance_cfg
.id)
3135 delta
["deleted"].append(instance_cfg
.id)
3140 def update_nsr_nsd(nsr_id
, xact
, scratch
):
3143 def get_nsr_vl_delta(nsr_id
, xact
, scratch
):
3144 delta
= {"added": [], "deleted": []}
3145 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3146 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3147 if elem_nsr_id
!= nsr_id
:
3150 if 'vld' in instance_cfg
.nsd
:
3151 for vld
in instance_cfg
.nsd
.vld
:
3152 delta
["added"].append(vld
)
3154 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3155 self
._log
.debug("NSR update: %s", instance_cfg
)
3156 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3157 if elem_nsr_id
!= nsr_id
:
3160 if 'vld' in instance_cfg
.nsd
:
3161 for vld
in instance_cfg
.nsd
.vld
:
3162 if vld
in delta
["added"]:
3163 delta
["added"].remove(vld
)
3165 delta
["deleted"].append(vld
)
3169 vl_delta
= yield from get_nsr_vl_delta(nsr_id
, xact
, scratch
)
3170 self
._log
.debug("Got NSR:%s VL instance delta: %s", nsr_id
, vl_delta
)
3172 for vld
in vl_delta
["added"]:
3173 yield from self
._nsm
.nsr_instantiate_vl(nsr_id
, vld
)
3175 for vld
in vl_delta
["deleted"]:
3176 yield from self
._nsm
.nsr_terminate_vl(nsr_id
, vld
)
3178 def get_add_delete_update_cfgs(dts_member_reg
, xact
, key_name
, scratch
):
3179 # Unfortunately, it is currently difficult to figure out what has exactly
3180 # changed in this xact without Pbdelta support (RIFT-4916)
3181 # As a workaround, we can fetch the pre and post xact elements and
3182 # perform a comparison to figure out adds/deletes/updates
3183 xact_cfgs
= list(dts_member_reg
.get_xact_elements(xact
))
3184 curr_cfgs
= list(dts_member_reg
.elements
)
3186 xact_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in xact_cfgs
}
3187 curr_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in curr_cfgs
}
3190 added_keys
= set(xact_key_map
) - set(curr_key_map
)
3191 added_cfgs
= [xact_key_map
[key
] for key
in added_keys
]
3194 deleted_keys
= set(curr_key_map
) - set(xact_key_map
)
3195 deleted_cfgs
= [curr_key_map
[key
] for key
in deleted_keys
]
3198 updated_keys
= set(curr_key_map
) & set(xact_key_map
)
3199 updated_cfgs
= [xact_key_map
[key
] for key
in updated_keys
3200 if xact_key_map
[key
] != curr_key_map
[key
]]
3202 return added_cfgs
, deleted_cfgs
, updated_cfgs
3204 def get_nsr_key_pairs(dts_member_reg
, xact
):
3206 for instance_cfg
, keyspec
in dts_member_reg
.get_xact_elements(xact
, include_keyspec
=True):
3207 self
._log
.debug("Key pair received is {} KS: {}".format(instance_cfg
, keyspec
))
3208 xpath
= keyspec
.to_xpath(RwNsrYang
.get_schema())
3209 key_pairs
[instance_cfg
.name
] = instance_cfg
3212 def on_apply(dts
, acg
, xact
, action
, scratch
):
3213 """Apply the configuration"""
3214 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3215 xact
, action
, scratch
)
3217 def handle_create_nsr(msg
, key_pairs
=None, restart_mode
=False):
3218 # Handle create nsr requests """
3219 # Do some validations
3220 if not msg
.has_field("nsd"):
3221 err
= "NSD not provided"
3222 self
._log
.error(err
)
3223 raise NetworkServiceRecordError(err
)
3225 self
._log
.debug("Creating NetworkServiceRecord %s from nsr config %s",
3226 msg
.id, msg
.as_dict())
3227 nsr
= self
.nsm
.create_nsr(msg
, key_pairs
=key_pairs
, restart_mode
=restart_mode
)
3230 def handle_delete_nsr(msg
):
3232 def delete_instantiation(ns_id
):
3233 """ Delete instantiation """
3234 with self
._dts
.transaction() as xact
:
3235 yield from self
._nsm
.terminate_ns(ns_id
, xact
)
3237 # Handle delete NSR requests
3238 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3239 # Terminate the NSR instance
3240 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3242 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3243 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3244 nsr
.record_event("terminate-rcvd", event_descr
)
3246 self
._loop
.create_task(delete_instantiation(msg
.id))
3249 def begin_instantiation(nsr
):
3250 # Begin instantiation
3251 self
._log
.info("Beginning NS instantiation: %s", nsr
.id)
3252 yield from self
._nsm
.instantiate_ns(nsr
.id, xact
)
3254 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3255 xact
, action
, scratch
)
3257 if action
== rwdts
.AppconfAction
.INSTALL
and xact
.id is None:
3259 for element
in self
._key
_pair
_regh
.elements
:
3260 key_pairs
.append(element
)
3261 for element
in self
._nsr
_regh
.elements
:
3262 nsr
= handle_create_nsr(element
, key_pairs
, restart_mode
=True)
3263 self
._loop
.create_task(begin_instantiation(nsr
))
3266 (added_msgs
, deleted_msgs
, updated_msgs
) = get_add_delete_update_cfgs(self
._nsr
_regh
,
3270 self
._log
.debug("Added: %s, Deleted: %s, Updated: %s", added_msgs
,
3271 deleted_msgs
, updated_msgs
)
3273 for msg
in added_msgs
:
3274 if msg
.id not in self
._nsm
.nsrs
:
3275 self
._log
.info("Create NSR received in on_apply to instantiate NS:%s", msg
.id)
3276 key_pairs
= get_nsr_key_pairs(self
._key
_pair
_regh
, xact
)
3277 nsr
= handle_create_nsr(msg
,key_pairs
)
3278 self
._loop
.create_task(begin_instantiation(nsr
))
3280 for msg
in deleted_msgs
:
3281 self
._log
.info("Delete NSR received in on_apply to terminate NS:%s", msg
.id)
3283 handle_delete_nsr(msg
)
3285 self
._log
.exception("Failed to terminate NS:%s", msg
.id)
3287 for msg
in updated_msgs
:
3288 self
._log
.info("Update NSR received in on_apply: %s", msg
)
3290 self
._nsm
.nsr_update_cfg(msg
.id, msg
)
3293 self
._loop
.create_task(update_nsr_nsd(msg
.id, xact
, scratch
))
3295 for group
in msg
.scaling_group
:
3296 instance_delta
= get_scale_group_instance_delta(msg
.id, group
.scaling_group_name_ref
, xact
)
3297 self
._log
.debug("Got NSR:%s scale group instance delta: %s", msg
.id, instance_delta
)
3299 for instance_id
in instance_delta
["added"]:
3300 self
._nsm
.scale_nsr_out(msg
.id, group
.scaling_group_name_ref
, instance_id
, xact
)
3302 for instance_id
in instance_delta
["deleted"]:
3303 self
._nsm
.scale_nsr_in(msg
.id, group
.scaling_group_name_ref
, instance_id
)
3306 return RwTypes
.RwStatus
.SUCCESS
3309 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
3310 """ Prepare calllback from DTS for NSR """
3312 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3313 action
= xact_info
.query_action
3315 "Got Nsr prepare callback (xact: %s) (action: %s) (info: %s), %s:%s)",
3316 xact
, action
, xact_info
, xpath
, msg
3320 def delete_instantiation(ns_id
):
3321 """ Delete instantiation """
3322 yield from self
._nsm
.terminate_ns(ns_id
, None)
3324 def handle_delete_nsr():
3325 """ Handle delete NSR requests """
3326 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3327 # Terminate the NSR instance
3328 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3330 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3331 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3332 nsr
.record_event("terminate-rcvd", event_descr
)
3334 self
._loop
.create_task(delete_instantiation(msg
.id))
3336 fref
= ProtobufC
.FieldReference
.alloc()
3337 fref
.goto_whole_message(msg
.to_pbcm())
3339 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
, rwdts
.QueryAction
.DELETE
]:
3340 # if this is an NSR create
3341 if action
!= rwdts
.QueryAction
.DELETE
and msg
.id not in self
._nsm
.nsrs
:
3342 # Ensure the Cloud account/datacenter has been specified
3343 if not msg
.has_field("cloud_account") and not msg
.has_field("om_datacenter"):
3344 raise NsrInstantiationFailed("Cloud account or datacenter not specified in NSR")
3346 # Check if nsd is specified
3347 if not msg
.has_field("nsd"):
3348 raise NsrInstantiationFailed("NSD not specified in NSR")
3351 nsr
= self
._nsm
.nsrs
[msg
.id]
3353 if msg
.has_field("nsd"):
3354 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3355 raise NsrVlUpdateError("Unable to update VL when NSR not in running state")
3356 if 'vld' not in msg
.nsd
or len(msg
.nsd
.vld
) == 0:
3357 raise NsrVlUpdateError("NS config NSD should have atleast 1 VLD defined")
3359 if msg
.has_field("scaling_group"):
3360 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3361 raise ScalingOperationError("Unable to perform scaling action when NS is not in running state")
3363 if len(msg
.scaling_group
) > 1:
3364 raise ScalingOperationError("Only a single scaling group can be configured at a time")
3366 for group_msg
in msg
.scaling_group
:
3367 num_new_group_instances
= len(group_msg
.instance
)
3368 if num_new_group_instances
> 1:
3369 raise ScalingOperationError("Only a single scaling instance can be modified at a time")
3371 elif num_new_group_instances
== 1:
3372 scale_group
= nsr
.scaling_groups
[group_msg
.scaling_group_name_ref
]
3373 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
]:
3374 if len(scale_group
.instances
) == scale_group
.max_instance_count
:
3375 raise ScalingOperationError("Max instances for %s reached" % scale_group
)
3377 acg
.handle
.prepare_complete_ok(xact_info
.handle
)
3380 self
._log
.debug("Registering for NSR config using xpath: %s",
3381 NsrDtsHandler
.NSR_XPATH
)
3383 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
3384 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
3385 self
._nsr
_regh
= acg
.register(xpath
=NsrDtsHandler
.NSR_XPATH
,
3386 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3387 on_prepare
=on_prepare
)
3389 self
._scale
_regh
= acg
.register(
3390 xpath
=NsrDtsHandler
.SCALE_INSTANCE_XPATH
,
3391 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY| rwdts
.Flag
.CACHE
,
3394 self
._key
_pair
_regh
= acg
.register(
3395 xpath
=NsrDtsHandler
.KEY_PAIR_XPATH
,
3396 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3400 class NsrOpDataDtsHandler(object):
3401 """ The network service op data DTS handler """
3402 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
3404 def __init__(self
, dts
, log
, loop
, nsm
):
3413 """ Return the registration handle"""
3418 """ Return the NS manager instance """
3423 """ Register for Nsr op data publisher registration"""
3424 self
._log
.debug("Registering Nsr op data path %s as publisher",
3425 NsrOpDataDtsHandler
.XPATH
)
3427 hdl
= rift
.tasklets
.DTS
.RegistrationHandler()
3428 handlers
= rift
.tasklets
.Group
.Handler()
3429 with self
._dts
.group_create(handler
=handlers
) as group
:
3430 self
._regh
= group
.register(xpath
=NsrOpDataDtsHandler
.XPATH
,
3432 flags
=rwdts
.Flag
.PUBLISHER | rwdts
.Flag
.NO_PREP_READ | rwdts
.Flag
.DATASTORE
)
3435 def create(self
, path
, msg
):
3437 Create an NS record in DTS with the path and message
3439 self
._log
.debug("Creating NSR %s:%s", path
, msg
)
3440 self
.regh
.create_element(path
, msg
)
3441 self
._log
.debug("Created NSR, %s:%s", path
, msg
)
3444 def update(self
, path
, msg
, flags
=rwdts
.XactFlag
.REPLACE
):
3446 Update an NS record in DTS with the path and message
3448 self
._log
.debug("Updating NSR, %s:%s regh = %s", path
, msg
, self
.regh
)
3449 self
.regh
.update_element(path
, msg
, flags
)
3450 self
._log
.debug("Updated NSR, %s:%s", path
, msg
)
3453 def delete(self
, path
):
3455 Update an NS record in DTS with the path and message
3457 self
._log
.debug("Deleting NSR path:%s", path
)
3458 self
.regh
.delete_element(path
)
3459 self
._log
.debug("Deleted NSR path:%s", path
)
3462 class VnfrDtsHandler(object):
3463 """ The virtual network service DTS handler """
3464 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
3466 def __init__(self
, dts
, log
, loop
, nsm
):
3476 """ Return registration handle """
3481 """ Return the NS manager instance """
3486 """ Register for vnfr create/update/delete/ advises from dts """
3488 def on_commit(xact_info
):
3489 """ The transaction has been committed """
3490 self
._log
.debug("Got vnfr commit (xact_info: %s)", xact_info
)
3491 return rwdts
.MemberRspCode
.ACTION_OK
3494 def on_prepare(xact_info
, action
, ks_path
, msg
):
3495 """ prepare callback from dts """
3496 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3498 "Got vnfr on_prepare cb (xact_info: %s, action: %s): %s:%s",
3499 xact_info
, action
, ks_path
, msg
3502 schema
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr
.schema()
3503 path_entry
= schema
.keyspec_to_entry(ks_path
)
3504 if path_entry
.key00
.id not in self
._nsm
._vnfrs
:
3505 self
._log
.error("%s request for non existent record path %s",
3507 xact_info
.respond_xpath(rwdts
.XactRspCode
.NA
, xpath
)
3511 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3512 if action
== rwdts
.QueryAction
.CREATE
or action
== rwdts
.QueryAction
.UPDATE
:
3513 yield from self
._nsm
.update_vnfr(msg
)
3514 elif action
== rwdts
.QueryAction
.DELETE
:
3515 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3516 self
._nsm
.delete_vnfr(path_entry
.key00
.id)
3518 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
, xpath
)
3520 self
._log
.debug("Registering for VNFR using xpath: %s",
3521 VnfrDtsHandler
.XPATH
,)
3523 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_commit
=on_commit
,
3524 on_prepare
=on_prepare
,)
3525 with self
._dts
.group_create() as group
:
3526 self
._regh
= group
.register(xpath
=VnfrDtsHandler
.XPATH
,
3528 flags
=(rwdts
.Flag
.SUBSCRIBER
),)
3531 class NsdRefCountDtsHandler(object):
3532 """ The NSD Ref Count DTS handler """
3533 XPATH
= "D,/nsr:ns-instance-opdata/rw-nsr:nsd-ref-count"
3535 def __init__(self
, dts
, log
, loop
, nsm
):
3545 """ Return registration handle """
3550 """ Return the NS manager instance """
3555 """ Register for NSD ref count read from dts """
3558 def on_prepare(xact_info
, action
, ks_path
, msg
):
3559 """ prepare callback from dts """
3560 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3562 if action
== rwdts
.QueryAction
.READ
:
3563 schema
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount
.schema()
3564 path_entry
= schema
.keyspec_to_entry(ks_path
)
3565 nsd_list
= yield from self
._nsm
.get_nsd_refcount(path_entry
.key00
.nsd_id_ref
)
3566 for xpath
, msg
in nsd_list
:
3567 xact_info
.respond_xpath(rsp_code
=rwdts
.XactRspCode
.MORE
,
3570 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
3572 raise NetworkServiceRecordError("Not supported operation %s" % action
)
3574 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_prepare
,)
3575 with self
._dts
.group_create() as group
:
3576 self
._regh
= group
.register(xpath
=NsdRefCountDtsHandler
.XPATH
,
3578 flags
=rwdts
.Flag
.PUBLISHER
,)
3581 class NsManager(object):
3582 """ The Network Service Manager class"""
3583 def __init__(self
, dts
, log
, loop
,
3584 nsr_handler
, vnfr_handler
, vlr_handler
, ro_plugin_selector
,
3585 vnffgmgr
, vnfd_pub_handler
, cloud_account_handler
):
3589 self
._nsr
_handler
= nsr_handler
3590 self
._vnfr
_pub
_handler
= vnfr_handler
3591 self
._vlr
_pub
_handler
= vlr_handler
3592 self
._vnffgmgr
= vnffgmgr
3593 self
._vnfd
_pub
_handler
= vnfd_pub_handler
3594 self
._cloud
_account
_handler
= cloud_account_handler
3596 self
._ro
_plugin
_selector
= ro_plugin_selector
3597 self
._ncclient
= rift
.mano
.ncclient
.NcClient(
3609 self
.cfgmgr_obj
= conman
.ROConfigManager(log
, loop
, dts
, self
)
3611 # TODO: All these handlers should move to tasklet level.
3612 # Passing self is often an indication of bad design
3613 self
._nsd
_dts
_handler
= NsdDtsHandler(dts
, log
, loop
, self
)
3614 self
._vnfd
_dts
_handler
= VnfdDtsHandler(dts
, log
, loop
, self
)
3615 self
._dts
_handlers
= [self
._nsd
_dts
_handler
,
3616 VnfrDtsHandler(dts
, log
, loop
, self
),
3617 NsdRefCountDtsHandler(dts
, log
, loop
, self
),
3618 NsrDtsHandler(dts
, log
, loop
, self
),
3619 ScalingRpcHandler(log
, dts
, loop
, self
.scale_rpc_callback
),
3620 NsrRpcDtsHandler(dts
,log
,loop
,self
),
3621 self
._vnfd
_dts
_handler
,
3642 def nsr_handler(self
):
3643 """" NSR handler """
3644 return self
._nsr
_handler
3648 """" So Obj handler """
3653 """ NSRs in this NSM"""
3658 """ NSDs in this NSM"""
3663 """ VNFDs in this NSM"""
3668 """ VNFRs in this NSM"""
3672 def nsr_pub_handler(self
):
3673 """ NSR publication handler """
3674 return self
._nsr
_handler
3677 def vnfr_pub_handler(self
):
3678 """ VNFR publication handler """
3679 return self
._vnfr
_pub
_handler
3682 def vlr_pub_handler(self
):
3683 """ VLR publication handler """
3684 return self
._vlr
_pub
_handler
3687 def vnfd_pub_handler(self
):
3688 return self
._vnfd
_pub
_handler
3692 """ Register all static DTS handlers """
3693 for dts_handle
in self
._dts
_handlers
:
3694 yield from dts_handle
.register()
3697 def get_ns_by_nsr_id(self
, nsr_id
):
3698 """ get NSR by nsr id """
3699 if nsr_id
not in self
._nsrs
:
3700 raise NetworkServiceRecordError("NSR id %s not found" % nsr_id
)
3702 return self
._nsrs
[nsr_id
]
3704 def scale_nsr_out(self
, nsr_id
, scale_group_name
, instance_id
, config_xact
):
3705 self
.log
.debug("Scale out NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3710 nsr
= self
._nsrs
[nsr_id
]
3711 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3712 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3714 self
._loop
.create_task(nsr
.create_scale_group_instance(scale_group_name
, instance_id
, config_xact
))
3716 def scale_nsr_in(self
, nsr_id
, scale_group_name
, instance_id
):
3717 self
.log
.debug("Scale in NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3722 nsr
= self
._nsrs
[nsr_id
]
3723 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3724 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3726 self
._loop
.create_task(nsr
.delete_scale_group_instance(scale_group_name
, instance_id
))
3728 def scale_rpc_callback(self
, xact
, msg
, action
):
3729 """Callback handler for RPC calls
3731 xact : Transaction Handler
3733 action : Scaling Action
3735 ScalingGroupInstance
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup_Instance
3736 ScalingGroup
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
3738 xpath
= ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').format(
3740 instance
= ScalingGroupInstance
.from_dict({"id": msg
.instance_id
})
3743 def get_nsr_scaling_group():
3744 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3746 for result
in results
:
3747 res
= yield from result
3748 nsr_config
= res
.result
3750 for scaling_group
in nsr_config
.scaling_group
:
3751 if scaling_group
.scaling_group_name_ref
== msg
.scaling_group_name_ref
:
3754 scaling_group
= nsr_config
.scaling_group
.add()
3755 scaling_group
.scaling_group_name_ref
= msg
.scaling_group_name_ref
3757 return (nsr_config
, scaling_group
)
3760 def update_config(nsr_config
):
3761 xml
= self
._ncclient
.convert_to_xml(RwNsrYang
, nsr_config
)
3762 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
3763 yield from self
._ncclient
.connect()
3764 yield from self
._ncclient
.manager
.edit_config(target
="running", config
=xml
, default_operation
="replace")
3768 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3769 scaling_group
.instance
.append(instance
)
3770 yield from update_config(nsr_config
)
3774 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3775 scaling_group
.instance
.remove(instance
)
3776 yield from update_config(nsr_config
)
3778 if action
== ScalingRpcHandler
.ACTION
.SCALE_OUT
:
3779 self
._loop
.create_task(scale_out())
3781 self
._loop
.create_task(scale_in())
3783 # Opdata based calls, disabled for now!
3784 # if action == ScalingRpcHandler.ACTION.SCALE_OUT:
3785 # self.scale_nsr_out(
3787 # msg.scaling_group_name_ref,
3791 # self.scale_nsr_in(
3793 # msg.scaling_group_name_ref,
3796 def nsr_update_cfg(self
, nsr_id
, msg
):
3797 nsr
= self
._nsrs
[nsr_id
]
3798 nsr
.nsr_cfg_msg
= msg
3800 def nsr_instantiate_vl(self
, nsr_id
, vld
):
3801 self
.log
.debug("NSR {} create VL {}".format(nsr_id
, vld
))
3802 nsr
= self
._nsrs
[nsr_id
]
3803 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3804 raise NsrVlUpdateError("Cannot perform VL instantiate if NSR is not in running state")
3806 # Not calling in a separate task as this is called from a separate task
3807 yield from nsr
.create_vl_instance(vld
)
3809 def nsr_terminate_vl(self
, nsr_id
, vld
):
3810 self
.log
.debug("NSR {} delete VL {}".format(nsr_id
, vld
.id))
3811 nsr
= self
._nsrs
[nsr_id
]
3812 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3813 raise NsrVlUpdateError("Cannot perform VL terminate if NSR is not in running state")
3815 # Not calling in a separate task as this is called from a separate task
3816 yield from nsr
.delete_vl_instance(vld
)
3818 def create_nsr(self
, nsr_msg
, key_pairs
=None,restart_mode
=False):
3819 """ Create an NSR instance """
3820 if nsr_msg
.id in self
._nsrs
:
3821 msg
= "NSR id %s already exists" % nsr_msg
.id
3822 self
._log
.error(msg
)
3823 raise NetworkServiceRecordError(msg
)
3825 self
._log
.info("Create NetworkServiceRecord nsr id %s from nsd_id %s",
3829 nsm_plugin
= self
._ro
_plugin
_selector
.ro_plugin
3830 sdn_account_name
= self
._cloud
_account
_handler
.get_cloud_account_sdn_name(nsr_msg
.cloud_account
)
3832 nsr
= NetworkServiceRecord(self
._dts
,
3840 restart_mode
=restart_mode
3842 self
._nsrs
[nsr_msg
.id] = nsr
3843 nsm_plugin
.create_nsr(nsr_msg
, nsr_msg
.nsd
, key_pairs
)
3847 def delete_nsr(self
, nsr_id
):
3849 Delete NSR with the passed nsr id
3851 del self
._nsrs
[nsr_id
]
3854 def instantiate_ns(self
, nsr_id
, config_xact
):
3855 """ Instantiate an NS instance """
3856 self
._log
.debug("Instantiating Network service id %s", nsr_id
)
3857 if nsr_id
not in self
._nsrs
:
3858 err
= "NSR id %s not found " % nsr_id
3859 self
._log
.error(err
)
3860 raise NetworkServiceRecordError(err
)
3862 nsr
= self
._nsrs
[nsr_id
]
3863 yield from nsr
.nsm_plugin
.instantiate_ns(nsr
, config_xact
)
3866 def update_vnfr(self
, vnfr
):
3867 """Create/Update an VNFR """
3869 vnfr_state
= self
._vnfrs
[vnfr
.id].state
3870 self
._log
.debug("Updating VNFR with state %s: vnfr %s", vnfr_state
, vnfr
)
3872 yield from self
._vnfrs
[vnfr
.id].update_state(vnfr
)
3873 nsr
= self
.find_nsr_for_vnfr(vnfr
.id)
3874 yield from nsr
.update_state()
3876 def find_nsr_for_vnfr(self
, vnfr_id
):
3877 """ Find the NSR which )has the passed vnfr id"""
3878 for nsr
in list(self
.nsrs
.values()):
3879 for vnfr
in list(nsr
.vnfrs
.values()):
3880 if vnfr
.id == vnfr_id
:
3884 def delete_vnfr(self
, vnfr_id
):
3885 """ Delete VNFR with the passed id"""
3886 del self
._vnfrs
[vnfr_id
]
3888 def get_nsd_ref(self
, nsd_id
):
3889 """ Get network service descriptor for the passed nsd_id
3891 nsd
= self
.get_nsd(nsd_id
)
3896 def get_nsr_config(self
, nsd_id
):
3897 xpath
= "C,/nsr:ns-instance-config"
3898 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3900 for result
in results
:
3901 entry
= yield from result
3902 ns_instance_config
= entry
.result
3904 for nsr
in ns_instance_config
.nsr
:
3905 if nsr
.nsd
.id == nsd_id
:
3911 def nsd_unref_by_nsr_id(self
, nsr_id
):
3912 """ Unref the network service descriptor based on NSR id """
3913 self
._log
.debug("NSR Unref called for Nsr Id:%s", nsr_id
)
3914 if nsr_id
in self
._nsrs
:
3915 nsr
= self
._nsrs
[nsr_id
]
3918 nsd
= self
.get_nsd(nsr
.nsd_id
)
3919 self
._log
.debug("Releasing ref on NSD %s held by NSR %s - Curr %d",
3920 nsd
.id, nsr
.id, nsd
.ref_count
)
3922 except NetworkServiceDescriptorError
:
3923 # We store a copy of NSD in NSR and the NSD in nsd-catalog
3928 self
._log
.error("Cannot find NSR with id %s", nsr_id
)
3929 raise NetworkServiceDescriptorUnrefError("No NSR with id" % nsr_id
)
3932 def nsd_unref(self
, nsd_id
):
3933 """ Unref the network service descriptor associated with the id """
3934 nsd
= self
.get_nsd(nsd_id
)
3937 def get_nsd(self
, nsd_id
):
3938 """ Get network service descriptor for the passed nsd_id"""
3939 if nsd_id
not in self
._nsds
:
3940 self
._log
.error("Cannot find NSD id:%s", nsd_id
)
3941 raise NetworkServiceDescriptorError("Cannot find NSD id:%s", nsd_id
)
3943 return self
._nsds
[nsd_id
]
3945 def create_nsd(self
, nsd_msg
):
3946 """ Create a network service descriptor """
3947 self
._log
.debug("Create network service descriptor - %s", nsd_msg
)
3948 if nsd_msg
.id in self
._nsds
:
3949 self
._log
.error("Cannot create NSD %s -NSD ID already exists", nsd_msg
)
3950 raise NetworkServiceDescriptorError("NSD already exists-%s", nsd_msg
.id)
3952 nsd
= NetworkServiceDescriptor(
3959 self
._nsds
[nsd_msg
.id] = nsd
3963 def update_nsd(self
, nsd
):
3964 """ update the Network service descriptor """
3965 self
._log
.debug("Update network service descriptor - %s", nsd
)
3966 if nsd
.id not in self
._nsds
:
3967 self
._log
.debug("No NSD found - creating NSD id = %s", nsd
.id)
3968 self
.create_nsd(nsd
)
3970 self
._log
.debug("Updating NSD id = %s, nsd = %s", nsd
.id, nsd
)
3971 self
._nsds
[nsd
.id].update(nsd
)
3973 def delete_nsd(self
, nsd_id
):
3974 """ Delete the Network service descriptor with the passed id """
3975 self
._log
.debug("Deleting the network service descriptor - %s", nsd_id
)
3976 if nsd_id
not in self
._nsds
:
3977 self
._log
.debug("Delete NSD failed - cannot find nsd-id %s", nsd_id
)
3978 raise NetworkServiceDescriptorNotFound("Cannot find %s", nsd_id
)
3980 if nsd_id
not in self
._nsds
:
3981 self
._log
.debug("Cannot delete NSD id %s reference exists %s",
3983 self
._nsds
[nsd_id
].ref_count
)
3984 raise NetworkServiceDescriptorRefCountExists(
3985 "Cannot delete :%s, ref_count:%s",
3987 self
._nsds
[nsd_id
].ref_count
)
3989 del self
._nsds
[nsd_id
]
3991 def get_vnfd_config(self
, xact
):
3992 vnfd_dts_reg
= self
._vnfd
_dts
_handler
.regh
3993 for cfg
in vnfd_dts_reg
.get_xact_elements(xact
):
3994 if cfg
.id not in self
._vnfds
:
3995 self
.create_vnfd(cfg
)
3997 def get_vnfd(self
, vnfd_id
, xact
):
3998 """ Get virtual network function descriptor for the passed vnfd_id"""
3999 if vnfd_id
not in self
._vnfds
:
4000 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4001 self
.get_vnfd_config(xact
)
4003 if vnfd_id
not in self
._vnfds
:
4004 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4005 raise VnfDescriptorError("Cannot find VNFD id:%s", vnfd_id
)
4007 return self
._vnfds
[vnfd_id
]
4009 def create_vnfd(self
, vnfd
):
4010 """ Create a virtual network function descriptor """
4011 self
._log
.debug("Create virtual network function descriptor - %s", vnfd
)
4012 if vnfd
.id in self
._vnfds
:
4013 self
._log
.error("Cannot create VNFD %s -VNFD ID already exists", vnfd
)
4014 raise VnfDescriptorError("VNFD already exists-%s", vnfd
.id)
4016 self
._vnfds
[vnfd
.id] = vnfd
4017 return self
._vnfds
[vnfd
.id]
4019 def update_vnfd(self
, vnfd
):
4020 """ Update the virtual network function descriptor """
4021 self
._log
.debug("Update virtual network function descriptor- %s", vnfd
)
4024 if vnfd
.id not in self
._vnfds
:
4025 self
._log
.debug("No VNFD found - creating VNFD id = %s", vnfd
.id)
4026 self
.create_vnfd(vnfd
)
4028 self
._log
.debug("Updating VNFD id = %s, vnfd = %s", vnfd
.id, vnfd
)
4029 self
._vnfds
[vnfd
.id] = vnfd
4032 def delete_vnfd(self
, vnfd_id
):
4033 """ Delete the virtual network function descriptor with the passed id """
4034 self
._log
.debug("Deleting the virtual network function descriptor - %s", vnfd_id
)
4035 if vnfd_id
not in self
._vnfds
:
4036 self
._log
.debug("Delete VNFD failed - cannot find vnfd-id %s", vnfd_id
)
4037 raise VnfDescriptorError("Cannot find %s", vnfd_id
)
4039 del self
._vnfds
[vnfd_id
]
4041 def nsd_in_use(self
, nsd_id
):
4042 """ Is the NSD with the passed id in use """
4043 self
._log
.debug("Is this NSD in use - msg:%s", nsd_id
)
4044 if nsd_id
in self
._nsds
:
4045 return self
._nsds
[nsd_id
].in_use()
4049 def publish_nsr(self
, xact
, path
, msg
):
4050 """ Publish a NSR """
4051 self
._log
.debug("Publish NSR with path %s, msg %s",
4053 yield from self
.nsr_handler
.update(xact
, path
, msg
)
4056 def unpublish_nsr(self
, xact
, path
):
4057 """ Un Publish an NSR """
4058 self
._log
.debug("Publishing delete NSR with path %s", path
)
4059 yield from self
.nsr_handler
.delete(path
, xact
)
4061 def vnfr_is_ready(self
, vnfr_id
):
4062 """ VNFR with the id is ready """
4063 self
._log
.debug("VNFR id %s ready", vnfr_id
)
4064 if vnfr_id
not in self
._vnfds
:
4065 err
= "Did not find VNFR ID with id %s" % vnfr_id
4066 self
._log
.critical("err")
4067 raise VirtualNetworkFunctionRecordError(err
)
4068 self
._vnfrs
[vnfr_id
].is_ready()
4071 def get_nsd_refcount(self
, nsd_id
):
4072 """ Get the nsd_list from this NSM"""
4074 def nsd_refcount_xpath(nsd_id
):
4075 """ xpath for ref count entry """
4076 return (NsdRefCountDtsHandler
.XPATH
+
4077 "[rw-nsr:nsd-id-ref = '{}']").format(nsd_id
)
4080 if nsd_id
is None or nsd_id
== "":
4081 for nsd
in self
._nsds
.values():
4082 nsd_msg
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4083 nsd_msg
.nsd_id_ref
= nsd
.id
4084 nsd_msg
.instance_ref_count
= nsd
.ref_count
4085 nsd_list
.append((nsd_refcount_xpath(nsd
.id), nsd_msg
))
4086 elif nsd_id
in self
._nsds
:
4087 nsd_msg
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
4088 nsd_msg
.nsd_id_ref
= self
._nsds
[nsd_id
].id
4089 nsd_msg
.instance_ref_count
= self
._nsds
[nsd_id
].ref_count
4090 nsd_list
.append((nsd_refcount_xpath(nsd_id
), nsd_msg
))
4095 def terminate_ns(self
, nsr_id
, xact
):
4097 Terminate network service for the given NSR Id
4100 # Terminate the instances/networks assocaited with this nw service
4101 self
._log
.debug("Terminating the network service %s", nsr_id
)
4102 yield from self
._nsrs
[nsr_id
].terminate()
4105 yield from self
.nsd_unref_by_nsr_id(nsr_id
)
4107 # Unpublish the NSR record
4108 self
._log
.debug("Unpublishing the network service %s", nsr_id
)
4109 yield from self
._nsrs
[nsr_id
].unpublish(xact
)
4111 # Finaly delete the NS instance from this NS Manager
4112 self
._log
.debug("Deletng the network service %s", nsr_id
)
4113 self
.delete_nsr(nsr_id
)
4116 class NsmRecordsPublisherProxy(object):
4117 """ This class provides a publisher interface that allows plugin objects
4118 to publish NSR/VNFR/VLR"""
4120 def __init__(self
, dts
, log
, loop
, nsr_pub_hdlr
, vnfr_pub_hdlr
, vlr_pub_hdlr
):
4124 self
._nsr
_pub
_hdlr
= nsr_pub_hdlr
4125 self
._vlr
_pub
_hdlr
= vlr_pub_hdlr
4126 self
._vnfr
_pub
_hdlr
= vnfr_pub_hdlr
4129 def publish_nsr(self
, xact
, nsr
):
4130 """ Publish an NSR """
4131 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4132 return (yield from self
._nsr
_pub
_hdlr
.update(xact
, path
, nsr
))
4135 def unpublish_nsr(self
, xact
, nsr
):
4136 """ Unpublish an NSR """
4137 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4138 return (yield from self
._nsr
_pub
_hdlr
.delete(xact
, path
))
4141 def publish_vnfr(self
, xact
, vnfr
):
4142 """ Publish an VNFR """
4143 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4144 return (yield from self
._vnfr
_pub
_hdlr
.update(xact
, path
, vnfr
))
4147 def unpublish_vnfr(self
, xact
, vnfr
):
4148 """ Unpublish a VNFR """
4149 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4150 return (yield from self
._vnfr
_pub
_hdlr
.delete(xact
, path
))
4153 def publish_vlr(self
, xact
, vlr
):
4154 """ Publish a VLR """
4155 path
= VirtualLinkRecord
.vlr_xpath(vlr
)
4156 return (yield from self
._vlr
_pub
_hdlr
.update(xact
, path
, vlr
))
4159 def unpublish_vlr(self
, xact
, vlr
):
4160 """ Unpublish a VLR """
4161 path
= VirtualLinkRecord
.vlr_xpath(vlr
)
4162 return (yield from self
._vlr
_pub
_hdlr
.delete(xact
, path
))
4165 class ScalingRpcHandler(mano_dts
.DtsHandler
):
4166 """ The Network service Monitor DTS handler """
4167 SCALE_IN_INPUT_XPATH
= "I,/nsr:exec-scale-in"
4168 SCALE_IN_OUTPUT_XPATH
= "O,/nsr:exec-scale-in"
4170 SCALE_OUT_INPUT_XPATH
= "I,/nsr:exec-scale-out"
4171 SCALE_OUT_OUTPUT_XPATH
= "O,/nsr:exec-scale-out"
4173 ACTION
= Enum('ACTION', 'SCALE_IN SCALE_OUT')
4175 def __init__(self
, log
, dts
, loop
, callback
=None):
4176 super().__init
__(log
, dts
, loop
)
4177 self
.callback
= callback
4178 self
.last_instance_id
= defaultdict(int)
4184 def on_scale_in_prepare(xact_info
, action
, ks_path
, msg
):
4185 assert action
== rwdts
.QueryAction
.RPC
4189 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_IN
)
4191 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleIn
.from_dict({
4192 "instance_id": msg
.instance_id
})
4194 xact_info
.respond_xpath(
4195 rwdts
.XactRspCode
.ACK
,
4196 self
.__class
__.SCALE_IN_OUTPUT_XPATH
,
4199 except Exception as e
:
4200 self
.log
.exception(e
)
4201 xact_info
.respond_xpath(
4202 rwdts
.XactRspCode
.NACK
,
4203 self
.__class
__.SCALE_IN_OUTPUT_XPATH
)
4206 def on_scale_out_prepare(xact_info
, action
, ks_path
, msg
):
4207 assert action
== rwdts
.QueryAction
.RPC
4210 scaling_group
= msg
.scaling_group_name_ref
4211 if not msg
.instance_id
:
4212 last_instance_id
= self
.last_instance_id
[scale_group
]
4213 msg
.instance_id
= last_instance_id
+ 1
4214 self
.last_instance_id
[scale_group
] += 1
4217 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_OUT
)
4219 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleOut
.from_dict({
4220 "instance_id": msg
.instance_id
})
4222 xact_info
.respond_xpath(
4223 rwdts
.XactRspCode
.ACK
,
4224 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
,
4227 except Exception as e
:
4228 self
.log
.exception(e
)
4229 xact_info
.respond_xpath(
4230 rwdts
.XactRspCode
.NACK
,
4231 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
)
4233 scale_in_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4234 on_prepare
=on_scale_in_prepare
)
4235 scale_out_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4236 on_prepare
=on_scale_out_prepare
)
4238 with self
.dts
.group_create() as group
:
4240 xpath
=self
.__class
__.SCALE_IN_INPUT_XPATH
,
4241 handler
=scale_in_hdl
,
4242 flags
=rwdts
.Flag
.PUBLISHER
)
4244 xpath
=self
.__class
__.SCALE_OUT_INPUT_XPATH
,
4245 handler
=scale_out_hdl
,
4246 flags
=rwdts
.Flag
.PUBLISHER
)
4249 class NsmTasklet(rift
.tasklets
.Tasklet
):
4251 The network service manager tasklet
4253 def __init__(self
, *args
, **kwargs
):
4254 super(NsmTasklet
, self
).__init
__(*args
, **kwargs
)
4255 self
.rwlog
.set_category("rw-mano-log")
4256 self
.rwlog
.set_subcategory("nsm")
4261 self
._ro
_plugin
_selector
= None
4262 self
._vnffgmgr
= None
4264 self
._nsr
_handler
= None
4265 self
._vnfr
_pub
_handler
= None
4266 self
._vlr
_pub
_handler
= None
4267 self
._vnfd
_pub
_handler
= None
4268 self
._scale
_cfg
_handler
= None
4270 self
._records
_publisher
_proxy
= None
4273 """ The task start callback """
4274 super(NsmTasklet
, self
).start()
4275 self
.log
.info("Starting NsmTasklet")
4277 self
.log
.debug("Registering with dts")
4278 self
._dts
= rift
.tasklets
.DTS(self
.tasklet_info
,
4279 RwNsmYang
.get_schema(),
4281 self
.on_dts_state_change
)
4283 self
.log
.debug("Created DTS Api GI Object: %s", self
._dts
)
4289 print("Caught Exception in NSM stop:", sys
.exc_info()[0])
4292 def on_instance_started(self
):
4293 """ Task instance started callback """
4294 self
.log
.debug("Got instance started callback")
4298 """ Task init callback """
4299 self
.log
.debug("Got instance started callback")
4301 self
.log
.debug("creating config account handler")
4303 self
._nsr
_pub
_handler
= publisher
.NsrOpDataDtsHandler(self
._dts
, self
.log
, self
.loop
)
4304 yield from self
._nsr
_pub
_handler
.register()
4306 self
._vnfr
_pub
_handler
= publisher
.VnfrPublisherDtsHandler(self
._dts
, self
.log
, self
.loop
)
4307 yield from self
._vnfr
_pub
_handler
.register()
4309 self
._vlr
_pub
_handler
= publisher
.VlrPublisherDtsHandler(self
._dts
, self
.log
, self
.loop
)
4310 yield from self
._vlr
_pub
_handler
.register()
4312 manifest
= self
.tasklet_info
.get_pb_manifest()
4313 use_ssl
= manifest
.bootstrap_phase
.rwsecurity
.use_ssl
4314 ssl_cert
= manifest
.bootstrap_phase
.rwsecurity
.cert
4315 ssl_key
= manifest
.bootstrap_phase
.rwsecurity
.key
4317 self
._vnfd
_pub
_handler
= publisher
.VnfdPublisher(use_ssl
, ssl_cert
, ssl_key
, self
.loop
)
4319 self
._records
_publisher
_proxy
= NsmRecordsPublisherProxy(
4323 self
._nsr
_pub
_handler
,
4324 self
._vnfr
_pub
_handler
,
4325 self
._vlr
_pub
_handler
,
4328 # Register the NSM to receive the nsm plugin
4329 # when cloud account is configured
4330 self
._ro
_plugin
_selector
= cloud
.ROAccountPluginSelector(
4334 self
._records
_publisher
_proxy
,
4336 yield from self
._ro
_plugin
_selector
.register()
4338 self
._cloud
_account
_handler
= cloud
.CloudAccountConfigSubscriber(
4343 yield from self
._cloud
_account
_handler
.register()
4345 self
._vnffgmgr
= rwvnffgmgr
.VnffgMgr(self
._dts
, self
.log
, self
.log_hdl
, self
.loop
)
4346 yield from self
._vnffgmgr
.register()
4348 self
._nsm
= NsManager(
4352 self
._nsr
_pub
_handler
,
4353 self
._vnfr
_pub
_handler
,
4354 self
._vlr
_pub
_handler
,
4355 self
._ro
_plugin
_selector
,
4357 self
._vnfd
_pub
_handler
,
4358 self
._cloud
_account
_handler
4361 yield from self
._nsm
.register()
4365 """ Task run callback """
4369 def on_dts_state_change(self
, state
):
4370 """Take action according to current dts state to transition
4371 application into the corresponding application state
4374 state - current dts state
4377 rwdts
.State
.INIT
: rwdts
.State
.REGN_COMPLETE
,
4378 rwdts
.State
.CONFIG
: rwdts
.State
.RUN
,
4382 rwdts
.State
.INIT
: self
.init
,
4383 rwdts
.State
.RUN
: self
.run
,
4386 # Transition application to next state
4387 handler
= handlers
.get(state
, None)
4388 if handler
is not None:
4389 yield from handler()
4391 # Transition dts to next state
4392 next_state
= switch
.get(state
, None)
4393 if next_state
is not None:
4394 self
.log
.debug("Changing state to %s", next_state
)
4395 self
._dts
.handle
.set_state(next_state
)