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
521 """ path for this object """
522 return "D,/vlr:vlr-catalog/vlr:vlr[vlr:id = '{}']".format(self
._vlr
_id
)
531 """ Get NSR name for this VL """
536 """ Virtual Link Desciptor """
540 def assigned_subnet(self
):
541 """ Subnet assigned to this VL"""
542 return self
._assigned
_subnet
547 Get the name for this VLR.
548 VLR name is "nsr name:VLD name"
550 if self
.vld_msg
.vim_network_name
:
551 return self
.vld_msg
.vim_network_name
552 elif self
.vld_msg
.name
== "multisite":
553 # This is a temporary hack to identify manually provisioned inter-site network
554 return self
.vld_msg
.name
556 return self
._nsr
_name
+ "." + self
.vld_msg
.name
559 def cloud_account_name(self
):
560 """ Cloud account that this VLR should be created in """
561 return self
._cloud
_account
_name
564 def om_datacenter_name(self
):
565 """ Datacenter that this VLR should be created in """
566 return self
._om
_datacenter
_name
570 """ Get the VLR path from VLR """
571 return (VirtualLinkRecord
.XPATH
+ "[vlr:id = '{}']").format(vlr
.id)
579 def state(self
, value
):
580 """ VLR set state """
584 def prev_state(self
):
585 """ VLR previous state """
586 return self
._prev
_state
589 def prev_state(self
, value
):
590 """ VLR set previous state """
591 self
._prev
_state
= value
595 """ Virtual Link Record message for Creating VLR in VNS """
596 vld_fields
= ["short_name",
604 vld_copy_dict
= {k
: v
for k
, v
in self
.vld_msg
.as_dict().items()
607 vlr_dict
= {"id": self
._vlr
_id
,
608 "nsr_id_ref": self
._nsr
_id
,
609 "vld_ref": self
.vld_msg
.id,
611 "cloud_account": self
.cloud_account_name
,
612 "om_datacenter": self
.om_datacenter_name
,
615 if self
._ip
_profile
and self
._ip
_profile
.has_field('ip_profile_params'):
616 vlr_dict
['ip_profile_params' ] = self
._ip
_profile
.ip_profile_params
.as_dict()
618 vlr_dict
.update(vld_copy_dict
)
619 vlr
= RwVlrYang
.YangData_Vlr_VlrCatalog_Vlr
.from_dict(vlr_dict
)
622 def reset_id(self
, vlr_id
):
623 self
._vlr
_id
= vlr_id
625 def create_nsr_vlr_msg(self
, vnfrs
):
626 """ The VLR message"""
627 nsr_vlr
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_Vlr()
628 nsr_vlr
.vlr_ref
= self
._vlr
_id
629 nsr_vlr
.assigned_subnet
= self
.assigned_subnet
630 nsr_vlr
.cloud_account
= self
.cloud_account_name
631 nsr_vlr
.om_datacenter
= self
.om_datacenter_name
633 for conn
in self
.vld_msg
.vnfd_connection_point_ref
:
635 if (vnfr
.vnfd
.id == conn
.vnfd_id_ref
and
636 vnfr
.member_vnf_index
== conn
.member_vnf_index_ref
and
637 self
.cloud_account_name
== vnfr
.cloud_account_name
and
638 self
.om_datacenter_name
== vnfr
.om_datacenter_name
):
639 cp_entry
= nsr_vlr
.vnfr_connection_point_ref
.add()
640 cp_entry
.vnfr_id
= vnfr
.id
641 cp_entry
.connection_point
= conn
.vnfd_connection_point_ref
646 def instantiate(self
):
647 """ Instantiate this VL """
649 self
._log
.debug("Instaniating VLR key %s, vld %s",
650 self
.xpath
, self
._vld
_msg
)
652 self
._state
= VlRecordState
.INSTANTIATION_PENDING
653 self
._log
.debug("Executing VL create path:%s msg:%s",
654 self
.xpath
, self
.vlr_msg
)
656 with self
._dts
.transaction(flags
=0) as xact
:
657 block
= xact
.block_create()
658 block
.add_query_create(self
.xpath
, self
.vlr_msg
)
659 self
._log
.debug("Executing VL create path:%s msg:%s",
660 self
.xpath
, self
.vlr_msg
)
661 res_iter
= yield from block
.execute(now
=True)
667 self
._state
= VlRecordState
.FAILED
668 raise NsrInstantiationFailed("Failed NS %s instantiation due to empty response" % self
.id)
670 if vlr
.operational_status
== 'failed':
671 self
._log
.debug("NS Id:%s VL creation failed for vlr id %s", self
.id, vlr
.id)
672 self
._state
= VlRecordState
.FAILED
673 raise NsrInstantiationFailed("Failed VL %s instantiation (%s)" % (vlr
.id, vlr
.operational_status_details
))
675 self
._log
.info("Instantiated VL with xpath %s and vlr:%s",
677 self
._state
= VlRecordState
.ACTIVE
678 self
._assigned
_subnet
= vlr
.assigned_subnet
680 def vlr_in_vns(self
):
681 """ Is there a VLR record in VNS """
682 if (self
._state
== VlRecordState
.ACTIVE
or
683 self
._state
== VlRecordState
.INSTANTIATION_PENDING
or
684 self
._state
== VlRecordState
.TERMINATE_PENDING
or
685 self
._state
== VlRecordState
.FAILED
):
692 """ Terminate this VL """
693 if not self
.vlr_in_vns():
694 self
._log
.debug("Ignoring terminate request for id %s in state %s",
695 self
.id, self
._state
)
698 self
._log
.debug("Terminating VL id:%s", self
.id)
699 self
._state
= VlRecordState
.TERMINATE_PENDING
701 with self
._dts
.transaction(flags
=0) as xact
:
702 block
= xact
.block_create()
703 block
.add_query_delete(self
.xpath
)
704 yield from block
.execute(flags
=0, now
=True)
706 self
._state
= VlRecordState
.TERMINATED
707 self
._log
.debug("Terminated VL id:%s", self
.id)
710 class VnfRecordState(Enum
):
711 """ Vnf Record State """
713 INSTANTIATION_PENDING
= 102
715 TERMINATE_PENDING
= 104
720 class VirtualNetworkFunctionRecord(object):
721 """ Virtual Network Function Record class"""
722 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
726 def create_record(dts
, log
, loop
, vnfd
, const_vnfd_msg
, nsd_id
, nsr_name
,
727 cloud_account_name
, om_datacenter_name
, nsr_id
, group_name
, group_instance_id
,
728 placement_groups
, restart_mode
=False):
729 """Creates a new VNFR object based on the given data.
731 If restart mode is enabled, then we look for existing records in the
732 DTS and create a VNFR records using the exiting data(ID)
735 VirtualNetworkFunctionRecord
737 vnfr_obj
= VirtualNetworkFunctionRecord(
751 restart_mode
=restart_mode
)
754 res_iter
= yield from dts
.query_read(
755 "D,/vnfr:vnfr-catalog/vnfr:vnfr",
756 rwdts
.XactFlag
.MERGE
)
759 response
= yield from fut
760 vnfr
= response
.result
762 if vnfr
.name
== vnfr_obj
.name
:
763 vnfr_obj
.reset_id(vnfr
.id)
780 group_instance_id
=None,
781 placement_groups
= [],
782 restart_mode
= False):
787 self
._const
_vnfd
_msg
= const_vnfd_msg
788 self
._nsd
_id
= nsd_id
789 self
._nsr
_name
= nsr_name
790 self
._nsr
_id
= nsr_id
791 self
._cloud
_account
_name
= cloud_account_name
792 self
._om
_datacenter
_name
= om_datacenter_name
793 self
._group
_name
= group_name
794 self
._group
_instance
_id
= group_instance_id
795 self
._placement
_groups
= placement_groups
796 self
._config
_status
= NsrYang
.ConfigStates
.INIT
798 self
._prev
_state
= VnfRecordState
.INIT
799 self
._state
= VnfRecordState
.INIT
800 self
._state
_failed
_reason
= None
802 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
805 self
._vnfr
_id
= str(uuid
.uuid4())
807 self
._vnfr
_msg
= self
.create_vnfr_msg()
808 self
._log
.debug("Set VNFR {} config type to {}".
809 format(self
.name
, self
.config_type
))
810 self
.restart_mode
= restart_mode
813 if group_name
is None and group_instance_id
is not None:
814 raise ValueError("Group instance id must not be provided with an empty group name")
824 return "D,/vnfr:vnfr-catalog/vnfr:vnfr[vnfr:id = '{}']".format(self
.id)
829 return self
._vnfr
_msg
832 def const_vnfr_msg(self
):
834 return RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_ConstituentVnfrRef(vnfr_id
=self
.id,cloud_account
=self
.cloud_account_name
,om_datacenter
=self
._om
_datacenter
_name
)
842 def cloud_account_name(self
):
843 """ Cloud account that this VNF should be created in """
844 return self
._cloud
_account
_name
847 def om_datacenter_name(self
):
848 """ Datacenter that this VNF should be created in """
849 return self
._om
_datacenter
_name
854 """ Is this VNF actve """
855 return True if self
._state
== VnfRecordState
.ACTIVE
else False
859 """ state of this VNF """
863 def state_failed_reason(self
):
864 """ Error message in case this VNF is in failed state """
865 return self
._state
_failed
_reason
868 def member_vnf_index(self
):
869 """ Member VNF index """
870 return self
._const
_vnfd
_msg
.member_vnf_index
875 return self
._nsr
_name
879 """ Name of this VNFR """
880 if self
._name
is not None:
883 name_tags
= [self
._nsr
_name
]
885 if self
._group
_name
is not None:
886 name_tags
.append(self
._group
_name
)
888 if self
._group
_instance
_id
is not None:
889 name_tags
.append(str(self
._group
_instance
_id
))
891 name_tags
.extend([self
.vnfd
.name
, str(self
.member_vnf_index
)])
893 self
._name
= "__".join(name_tags
)
898 def vnfr_xpath(vnfr
):
899 """ Get the VNFR path from VNFR """
900 return (VirtualNetworkFunctionRecord
.XPATH
+ "[vnfr:id = '{}']").format(vnfr
.id)
903 def config_type(self
):
904 cfg_types
= ['netconf', 'juju', 'script']
905 for method
in cfg_types
:
906 if self
._vnfd
.vnf_configuration
.has_field(method
):
911 def config_status(self
):
912 """Return the config status as YANG ENUM string"""
913 self
._log
.debug("Map VNFR {} config status {} ({})".
914 format(self
.name
, self
._config
_status
, self
.config_type
))
915 if self
.config_type
== 'none':
916 return 'config_not_needed'
917 elif self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
919 elif self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
924 def set_state(self
, state
):
925 """ set the state of this object """
926 self
._prev
_state
= self
._state
929 def reset_id(self
, vnfr_id
):
930 self
._vnfr
_id
= vnfr_id
931 self
._vnfr
_msg
= self
.create_vnfr_msg()
934 self
.config_store
.merge_vnfd_config(
937 self
.member_vnf_index
,
940 def create_vnfr_msg(self
):
941 """ VNFR message for this VNFR """
949 vnfd_copy_dict
= {k
: v
for k
, v
in self
._vnfd
.as_dict().items() if k
in vnfd_fields
}
952 "nsr_id_ref": self
._nsr
_id
,
953 "vnfd_ref": self
.vnfd
.id,
955 "cloud_account": self
._cloud
_account
_name
,
956 "om_datacenter": self
._om
_datacenter
_name
,
957 "config_status": self
.config_status
959 vnfr_dict
.update(vnfd_copy_dict
)
961 vnfr
= RwVnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr
.from_dict(vnfr_dict
)
962 vnfr
.member_vnf_index_ref
= self
.member_vnf_index
963 vnfr
.vnf_configuration
.from_dict(self
._vnfd
.vnf_configuration
.as_dict())
965 if self
._vnfd
.mgmt_interface
.has_field("port"):
966 vnfr
.mgmt_interface
.port
= self
._vnfd
.mgmt_interface
.port
968 for group_info
in self
._placement
_groups
:
969 group
= vnfr
.placement_groups_info
.add()
970 group
.from_dict(group_info
.as_dict())
972 # UI expects the monitoring param field to exist
973 vnfr
.monitoring_param
= []
975 self
._log
.debug("Get vnfr_msg for VNFR {} : {}".format(self
.name
, vnfr
))
979 def update_vnfm(self
):
980 self
._log
.debug("Send an update to VNFM for VNFR {} with {}".
981 format(self
.name
, self
.vnfr_msg
))
982 yield from self
._dts
.query_update(
984 rwdts
.XactFlag
.TRACE
,
988 def get_config_status(self
):
989 """Return the config status as YANG ENUM"""
990 return self
._config
_status
993 def set_config_status(self
, status
):
995 def status_to_string(status
):
997 NsrYang
.ConfigStates
.INIT
: 'init',
998 NsrYang
.ConfigStates
.CONFIGURING
: 'configuring',
999 NsrYang
.ConfigStates
.CONFIG_NOT_NEEDED
: 'config_not_needed',
1000 NsrYang
.ConfigStates
.CONFIGURED
: 'configured',
1001 NsrYang
.ConfigStates
.FAILED
: 'failed',
1004 return status_dc
[status
]
1006 self
._log
.debug("Update VNFR {} from {} ({}) to {}".
1007 format(self
.name
, self
._config
_status
,
1008 self
.config_type
, status
))
1009 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1010 self
._log
.error("Updating already configured VNFR {}".
1014 if self
._config
_status
!= status
:
1016 self
._config
_status
= status
1017 # I don't think this is used. Original implementor can check.
1018 # Caused Exception, so corrected it by status_to_string
1019 # But not sure whats the use of this variable?
1020 self
.vnfr_msg
.config_status
= status_to_string(status
)
1021 except Exception as e
:
1022 self
._log
.error("Exception=%s", str(e
))
1025 self
._log
.debug("Updated VNFR {} status to {}".format(self
.name
, status
))
1027 if self
._config
_status
!= NsrYang
.ConfigStates
.INIT
:
1029 # Publish only after VNFM has the VNFR created
1030 yield from self
.update_vnfm()
1031 except Exception as e
:
1032 self
._log
.error("Exception updating VNFM with new status {} of VNFR {}: {}".
1033 format(status
, self
.name
, e
))
1034 self
._log
.exception(e
)
1036 def is_configured(self
):
1037 if self
.config_type
== 'none':
1040 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1046 def instantiate(self
, nsr
):
1047 """ Instantiate this VNFR"""
1049 self
._log
.debug("Instaniating VNFR key %s, vnfd %s",
1050 self
.xpath
, self
._vnfd
)
1052 self
._log
.debug("Create VNF with xpath %s and vnfr %s",
1053 self
.xpath
, self
.vnfr_msg
)
1055 self
.set_state(VnfRecordState
.INSTANTIATION_PENDING
)
1057 def find_vlr_for_cp(conn
):
1058 """ Find VLR for the given connection point """
1059 for vlr
in nsr
.vlrs
:
1060 for vnfd_cp
in vlr
.vld_msg
.vnfd_connection_point_ref
:
1061 if (vnfd_cp
.vnfd_id_ref
== self
._vnfd
.id and
1062 vnfd_cp
.vnfd_connection_point_ref
== conn
.name
and
1063 vnfd_cp
.member_vnf_index_ref
== self
.member_vnf_index
and
1064 vlr
.cloud_account_name
== self
.cloud_account_name
):
1065 self
._log
.debug("Found VLR for cp_name:%s and vnf-index:%d",
1066 conn
.name
, self
.member_vnf_index
)
1070 # For every connection point in the VNFD fill in the identifier
1071 for conn_p
in self
._vnfd
.connection_point
:
1072 cpr
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_ConnectionPoint()
1073 cpr
.name
= conn_p
.name
1074 cpr
.type_yang
= conn_p
.type_yang
1075 vlr_ref
= find_vlr_for_cp(conn_p
)
1077 msg
= "Failed to find VLR for cp = %s" % conn_p
.name
1078 self
._log
.debug("%s", msg
)
1079 # raise VirtualNetworkFunctionRecordError(msg)
1082 cpr
.vlr_ref
= vlr_ref
.id
1083 self
.vnfr_msg
.connection_point
.append(cpr
)
1084 self
._log
.debug("Connection point [%s] added, vnf id=%s vnfd id=%s",
1085 cpr
, self
.vnfr_msg
.id, self
.vnfr_msg
.vnfd_ref
)
1087 if not self
.restart_mode
:
1088 yield from self
._dts
.query_create(self
.xpath
,
1092 yield from self
._dts
.query_update(self
.xpath
,
1096 self
._log
.info("Created VNF with xpath %s and vnfr %s",
1097 self
.xpath
, self
.vnfr_msg
)
1099 self
._log
.info("Instantiated VNFR with xpath %s and vnfd %s, vnfr %s",
1100 self
.xpath
, self
._vnfd
, self
.vnfr_msg
)
1103 def update_state(self
, vnfr_msg
):
1104 """ Update this VNFR"""
1105 if vnfr_msg
.operational_status
== "running":
1106 if self
.vnfr_msg
.operational_status
!= "running":
1107 yield from self
.is_active()
1108 elif vnfr_msg
.operational_status
== "failed":
1109 yield from self
.instantiation_failed(failed_reason
=vnfr_msg
.operational_status_details
)
1112 def is_active(self
):
1113 """ This VNFR is active """
1114 self
._log
.debug("VNFR %s is active", self
._vnfr
_id
)
1115 self
.set_state(VnfRecordState
.ACTIVE
)
1118 def instantiation_failed(self
, failed_reason
=None):
1119 """ This VNFR instantiation failed"""
1120 self
._log
.error("VNFR %s instantiation failed", self
._vnfr
_id
)
1121 self
.set_state(VnfRecordState
.FAILED
)
1122 self
._state
_failed
_reason
= failed_reason
1124 def vnfr_in_vnfm(self
):
1125 """ Is there a VNFR record in VNFM """
1126 if (self
._state
== VnfRecordState
.ACTIVE
or
1127 self
._state
== VnfRecordState
.INSTANTIATION_PENDING
or
1128 self
._state
== VnfRecordState
.FAILED
):
1134 def terminate(self
):
1135 """ Terminate this VNF """
1136 if not self
.vnfr_in_vnfm():
1137 self
._log
.debug("Ignoring terminate request for id %s in state %s",
1138 self
.id, self
._state
)
1141 self
._log
.debug("Terminating VNF id:%s", self
.id)
1142 self
.set_state(VnfRecordState
.TERMINATE_PENDING
)
1143 with self
._dts
.transaction(flags
=0) as xact
:
1144 block
= xact
.block_create()
1145 block
.add_query_delete(self
.xpath
)
1146 yield from block
.execute(flags
=0)
1147 self
.set_state(VnfRecordState
.TERMINATED
)
1148 self
._log
.debug("Terminated VNF id:%s", self
.id)
1151 class NetworkServiceStatus(object):
1152 """ A class representing the Network service's status """
1153 MAX_EVENTS_RECORDED
= 10
1154 """ Network service Status class"""
1155 def __init__(self
, dts
, log
, loop
):
1160 self
._state
= NetworkServiceRecordState
.INIT
1161 self
._events
= deque([])
1164 def create_notification(self
, evt
, evt_desc
, evt_details
):
1165 xp
= "N,/rw-nsr:nsm-notification"
1166 notif
= RwNsrYang
.YangNotif_RwNsr_NsmNotification()
1168 notif
.description
= evt_desc
1169 notif
.details
= evt_details
if evt_details
is not None else None
1171 yield from self
._dts
.query_create(xp
, rwdts
.XactFlag
.ADVISE
, notif
)
1172 self
._log
.info("Notification called by creating dts query: %s", notif
)
1174 def record_event(self
, evt
, evt_desc
, evt_details
):
1175 """ Record an event """
1176 self
._log
.debug("Recording event - evt %s, evt_descr %s len = %s",
1177 evt
, evt_desc
, len(self
._events
))
1178 if len(self
._events
) >= NetworkServiceStatus
.MAX_EVENTS_RECORDED
:
1179 self
._events
.popleft()
1180 self
._events
.append((int(time
.time()), evt
, evt_desc
,
1181 evt_details
if evt_details
is not None else None))
1183 self
._loop
.create_task(self
.create_notification(evt
,evt_desc
,evt_details
))
1185 def set_state(self
, state
):
1186 """ set the state of this status object """
1190 """ Return the state as a yang enum string """
1191 state_to_str_map
= {"INIT": "init",
1192 "VL_INIT_PHASE": "vl_init_phase",
1193 "VNF_INIT_PHASE": "vnf_init_phase",
1194 "VNFFG_INIT_PHASE": "vnffg_init_phase",
1195 "SCALING_GROUP_INIT_PHASE": "scaling_group_init_phase",
1196 "RUNNING": "running",
1197 "SCALING_OUT": "scaling_out",
1198 "SCALING_IN": "scaling_in",
1199 "TERMINATE_RCVD": "terminate_rcvd",
1200 "TERMINATE": "terminate",
1201 "VL_TERMINATE_PHASE": "vl_terminate_phase",
1202 "VNF_TERMINATE_PHASE": "vnf_terminate_phase",
1203 "VNFFG_TERMINATE_PHASE": "vnffg_terminate_phase",
1204 "TERMINATED": "terminated",
1206 "VL_INSTANTIATE": "vl_instantiate",
1207 "VL_TERMINATE": "vl_terminate",
1209 return state_to_str_map
[self
._state
.name
]
1213 """ State of this status object """
1218 """ Network Service Record as a message"""
1221 for entry
in self
._events
:
1222 event
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_OperationalEvents()
1225 event
.timestamp
, event
.event
, event
.description
, event
.details
= entry
1226 event_list
.append(event
)
1230 class NetworkServiceRecord(object):
1231 """ Network service record """
1232 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
1234 def __init__(self
, dts
, log
, loop
, nsm
, nsm_plugin
, nsr_cfg_msg
, sdn_account_name
, key_pairs
, restart_mode
=False):
1239 self
._nsr
_cfg
_msg
= nsr_cfg_msg
1240 self
._nsm
_plugin
= nsm_plugin
1241 self
._sdn
_account
_name
= sdn_account_name
1244 self
._nsr
_msg
= None
1245 self
._nsr
_regh
= None
1246 self
._key
_pairs
= key_pairs
1251 self
._param
_pools
= {}
1252 self
._scaling
_groups
= {}
1253 self
._create
_time
= int(time
.time())
1254 self
._op
_status
= NetworkServiceStatus(dts
, log
, loop
)
1255 self
._config
_status
= NsrYang
.ConfigStates
.CONFIGURING
1256 self
._config
_status
_details
= None
1258 self
.restart_mode
= restart_mode
1259 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
1260 self
._debug
_running
= False
1261 self
._is
_active
= False
1262 self
._vl
_phase
_completed
= False
1263 self
._vnf
_phase
_completed
= False
1266 # Initalise the state to init
1267 # The NSR moves through the following transitions
1268 # 1. INIT -> VLS_READY once all the VLs in the NSD are created
1269 # 2. VLS_READY - VNFS_READY when all the VNFs in the NSD are created
1270 # 3. VNFS_READY - READY when the NSR is published
1272 self
.set_state(NetworkServiceRecordState
.INIT
)
1274 self
.substitute_input_parameters
= InputParameterSubstitution(self
._log
)
1277 def nsm_plugin(self
):
1279 return self
._nsm
_plugin
1281 def set_state(self
, state
):
1282 """ Set state for this NSR"""
1283 self
._log
.debug("Setting state to %s", state
)
1284 # We are in init phase and is moving to the next state
1285 # The new state could be a FAILED state or VNF_INIIT_PHASE
1286 if self
.state
== NetworkServiceRecordState
.VL_INIT_PHASE
:
1287 self
._vl
_phase
_completed
= True
1289 if self
.state
== NetworkServiceRecordState
.VNF_INIT_PHASE
:
1290 self
._vnf
_phase
_completed
= True
1292 self
._op
_status
.set_state(state
)
1296 """ Get id for this NSR"""
1297 return self
._nsr
_cfg
_msg
.id
1301 """ Name of this network service record """
1302 return self
._nsr
_cfg
_msg
.name
1305 def cloud_account_name(self
):
1306 return self
._nsr
_cfg
_msg
.cloud_account
1309 def om_datacenter_name(self
):
1310 if self
._nsr
_cfg
_msg
.has_field('om_datacenter'):
1311 return self
._nsr
_cfg
_msg
.om_datacenter
1316 """State of this NetworkServiceRecord"""
1317 return self
._op
_status
.state
1321 """ Is this NSR active ?"""
1322 return True if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
else False
1326 """ VLRs associated with this NSR"""
1331 """ VNFRs associated with this NSR"""
1336 """ VNFFGRs associated with this NSR"""
1337 return self
._vnffgrs
1340 def scaling_groups(self
):
1341 """ Scaling groups associated with this NSR """
1342 return self
._scaling
_groups
1345 def param_pools(self
):
1346 """ Parameter value pools associated with this NSR"""
1347 return self
._param
_pools
1350 def nsr_cfg_msg(self
):
1351 return self
._nsr
_cfg
_msg
1354 def nsr_cfg_msg(self
, msg
):
1355 self
._nsr
_cfg
_msg
= msg
1359 """ NSD Protobuf for this NSR """
1360 if self
._nsd
is not None:
1362 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
1367 """ NSD ID for this NSR """
1368 return self
.nsd_msg
.id
1372 ''' Get a new job id for config primitive'''
1377 def config_status(self
):
1378 """ Config status for NSR """
1379 return self
._config
_status
1381 def resolve_placement_group_cloud_construct(self
, input_group
):
1383 Returns the cloud specific construct for placement group
1385 copy_dict
= ['name', 'requirement', 'strategy']
1387 for group_info
in self
._nsr
_cfg
_msg
.nsd_placement_group_maps
:
1388 if group_info
.placement_group_ref
== input_group
.name
:
1389 group
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr_PlacementGroupsInfo()
1390 group_dict
= {k
:v
for k
,v
in
1391 group_info
.as_dict().items() if k
!= 'placement_group_ref'}
1392 for param
in copy_dict
:
1393 group_dict
.update({param
: getattr(input_group
, param
)})
1394 group
.from_dict(group_dict
)
1400 return "NSR(name={}, nsd_id={}, cloud_account={})".format(
1401 self
.name
, self
.nsd_id
, self
.cloud_account_name
1404 def _get_vnfd(self
, vnfd_id
, config_xact
):
1405 """ Fetch vnfd msg for the passed vnfd id """
1406 return self
._nsm
.get_vnfd(vnfd_id
, config_xact
)
1408 def _get_vnfd_cloud_account(self
, vnfd_member_index
):
1409 """ Fetch Cloud Account for the passed vnfd id """
1410 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1411 vim_accounts
= [(vnf
.cloud_account
,vnf
.om_datacenter
) for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map \
1412 if vnfd_member_index
== vnf
.member_vnf_index_ref
]
1413 if vim_accounts
and vim_accounts
[0]:
1414 return vim_accounts
[0]
1415 return (self
.cloud_account_name
,self
.om_datacenter_name
)
1417 def _get_constituent_vnfd_msg(self
, vnf_index
):
1418 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1419 if const_vnfd
.member_vnf_index
== vnf_index
:
1422 raise ValueError("Constituent VNF index %s not found" % vnf_index
)
1424 def record_event(self
, evt
, evt_desc
, evt_details
=None, state
=None):
1425 """ Record an event """
1426 self
._op
_status
.record_event(evt
, evt_desc
, evt_details
)
1427 if state
is not None:
1428 self
.set_state(state
)
1430 def scaling_trigger_str(self
, trigger
):
1431 SCALING_TRIGGER_STRS
= {
1432 NsdYang
.ScalingTrigger
.PRE_SCALE_IN
: 'pre-scale-in',
1433 NsdYang
.ScalingTrigger
.POST_SCALE_IN
: 'post-scale-in',
1434 NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
: 'pre-scale-out',
1435 NsdYang
.ScalingTrigger
.POST_SCALE_OUT
: 'post-scale-out',
1438 return SCALING_TRIGGER_STRS
[trigger
]
1439 except Exception as e
:
1440 self
._log
.error("Scaling trigger mapping error for {} : {}".
1442 self
._log
.exception(e
)
1443 return "Unknown trigger"
1446 def instantiate_vls(self
):
1448 This function instantiates VLs for every VL in this Network Service
1450 self
._log
.debug("Instantiating %d VLs in NSD id %s", len(self
._vlrs
),
1452 for vlr
in self
._vlrs
:
1453 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1454 vlr
.state
= VlRecordState
.ACTIVE
1457 def create(self
, config_xact
):
1458 """ Create this network service"""
1459 # Create virtual links for all the external vnf
1460 # connection points in this NS
1461 yield from self
.create_vls()
1463 # Create VNFs in this network service
1464 yield from self
.create_vnfs(config_xact
)
1466 # Create VNFFG for network service
1467 self
.create_vnffgs()
1469 # Create Scaling Groups for each scaling group in NSD
1470 self
.create_scaling_groups()
1472 # Create Parameter Pools
1473 self
.create_param_pools()
1476 def apply_scale_group_config_script(self
, script
, group
, scale_instance
, trigger
, vnfrs
=None):
1477 """ Apply config based on script for scale group """
1480 def add_vnfrs_data(vnfrs_list
):
1481 """ Add as a dict each of the VNFRs data """
1483 for vnfr
in vnfrs_list
:
1484 self
._log
.debug("Add VNFR {} data".format(vnfr
))
1486 vnfr_data
['name'] = vnfr
.name
1487 if trigger
in [NsdYang
.ScalingTrigger
.PRE_SCALE_IN
, NsdYang
.ScalingTrigger
.POST_SCALE_OUT
]:
1488 # Get VNF management and other IPs, etc
1489 opdata
= yield from self
.fetch_vnfr(vnfr
.xpath
)
1490 self
._log
.debug("VNFR {} op data: {}".format(vnfr
.name
, opdata
))
1492 vnfr_data
['rw_mgmt_ip'] = opdata
.mgmt_interface
.ip_address
1493 vnfr_data
['rw_mgmt_port'] = opdata
.mgmt_interface
.port
1494 except Exception as e
:
1495 self
._log
.error("Unable to get management IP for vnfr {}:{}".
1496 format(vnfr
.name
, e
))
1499 vnfr_data
['connection_points'] = []
1500 for cp
in opdata
.connection_point
:
1502 con_pt
['name'] = cp
.name
1503 con_pt
['ip_address'] = cp
.ip_address
1504 vnfr_data
['connection_points'].append(con_pt
)
1505 except Exception as e
:
1506 self
._log
.error("Exception getting connections points for VNFR {}: {}".
1507 format(vnfr
.name
, e
))
1509 vnfrs_data
.append(vnfr_data
)
1510 self
._log
.debug("VNFRs data: {}".format(vnfrs_data
))
1514 def add_nsr_data(nsr
):
1516 nsr_data
['name'] = nsr
.name
1519 if script
is None or len(script
) == 0:
1520 self
._log
.error("Script not provided for scale group config: {}".format(group
.name
))
1523 if script
[0] == '/':
1526 path
= os
.path
.join(os
.environ
['RIFT_INSTALL'], "usr/bin", script
)
1527 if not os
.path
.exists(path
):
1528 self
._log
.error("Config faled for scale group {}: Script does not exist at {}".
1529 format(group
.name
, path
))
1532 # Build a YAML file with all parameters for the script to execute
1533 # The data consists of 5 sections
1535 # 2. Scale group config
1536 # 3. VNFRs in the scale group
1537 # 4. VNFRs outside scale group
1540 data
['trigger'] = group
.trigger_map(trigger
)
1541 data
['config'] = group
.group_msg
.as_dict()
1544 data
["vnfrs_in_group"] = yield from add_vnfrs_data(vnfrs
)
1546 data
["vnfrs_in_group"] = yield from add_vnfrs_data(scale_instance
.vnfrs
)
1548 data
["vnfrs_others"] = yield from add_vnfrs_data(self
.vnfrs
.values())
1549 data
["nsr"] = add_nsr_data(self
)
1552 with tempfile
.NamedTemporaryFile(delete
=False) as tmp_file
:
1553 tmp_file
.write(yaml
.dump(data
, default_flow_style
=True)
1556 self
._log
.debug("Creating a temp file: {} with input data: {}".
1557 format(tmp_file
.name
, data
))
1559 cmd
= "{} {}".format(path
, tmp_file
.name
)
1560 self
._log
.debug("Running the CMD: {}".format(cmd
))
1561 proc
= yield from asyncio
.create_subprocess_shell(cmd
, loop
=self
._loop
)
1562 rc
= yield from proc
.wait()
1564 self
._log
.error("The script {} for scale group {} config returned: {}".
1565 format(script
, group
.name
, rc
))
1573 def apply_scaling_group_config(self
, trigger
, group
, scale_instance
, vnfrs
=None):
1574 """ Apply the config for the scaling group based on trigger """
1575 if group
is None or scale_instance
is None:
1579 def update_config_status(success
=True, err_msg
=None):
1580 self
._log
.debug("Update %s config status to %r : %s",
1581 scale_instance
, success
, err_msg
)
1582 if (scale_instance
.config_status
== "failed"):
1583 # Do not update the config status if it is already in failed state
1586 if scale_instance
.config_status
== "configured":
1587 # Update only to failed state an already configured scale instance
1589 scale_instance
.config_status
= "failed"
1590 scale_instance
.config_err_msg
= err_msg
1591 yield from self
.update_state()
1593 # We are in configuring state
1594 # Only after post scale out mark instance as configured
1595 if trigger
== NsdYang
.ScalingTrigger
.POST_SCALE_OUT
:
1597 scale_instance
.config_status
= "configured"
1599 scale_instance
.config_status
= "failed"
1600 scale_instance
.config_err_msg
= err_msg
1601 yield from self
.update_state()
1603 config
= group
.trigger_config(trigger
)
1607 self
._log
.debug("Scaling group {} config: {}".format(group
.name
, config
))
1608 if config
.has_field("ns_config_primitive_name_ref"):
1609 config_name
= config
.ns_config_primitive_name_ref
1610 nsd_msg
= self
.nsd_msg
1611 config_primitive
= None
1612 for ns_cfg_prim
in nsd_msg
.service_primitive
:
1613 if ns_cfg_prim
.name
== config_name
:
1614 config_primitive
= ns_cfg_prim
1617 if config_primitive
is None:
1618 raise ValueError("Could not find ns_cfg_prim %s in nsr %s" % (config_name
, self
.name
))
1620 self
._log
.debug("Scaling group {} config primitive: {}".format(group
.name
, config_primitive
))
1621 if config_primitive
.has_field("user_defined_script"):
1622 rc
= yield from self
.apply_scale_group_config_script(config_primitive
.user_defined_script
,
1623 group
, scale_instance
, trigger
, vnfrs
)
1626 err_msg
= "Failed config for trigger {} using config script '{}'". \
1627 format(self
.scaling_trigger_str(trigger
),
1628 config_primitive
.user_defined_script
)
1629 yield from update_config_status(success
=rc
, err_msg
=err_msg
)
1632 err_msg
= "Failed config for trigger {} as config script is not specified". \
1633 format(self
.scaling_trigger_str(trigger
))
1634 yield from update_config_status(success
=False, err_msg
=err_msg
)
1635 raise NotImplementedError("Only script based config support for scale group for now: {}".
1638 err_msg
= "Failed config for trigger {} as config primitive is not specified".\
1639 format(self
.scaling_trigger_str(trigger
))
1640 yield from update_config_status(success
=False, err_msg
=err_msg
)
1641 self
._log
.error("Config primitive not specified for config action in scale group %s" %
1645 def create_scaling_groups(self
):
1646 """ This function creates a NSScalingGroup for every scaling
1647 group defined in he NSD"""
1649 for scaling_group_msg
in self
.nsd_msg
.scaling_group_descriptor
:
1650 self
._log
.debug("Found scaling_group %s in nsr id %s",
1651 scaling_group_msg
.name
, self
.id)
1653 group_record
= scale_group
.ScalingGroup(
1658 self
._scaling
_groups
[group_record
.name
] = group_record
1661 def create_scale_group_instance(self
, group_name
, index
, config_xact
, is_default
=False):
1662 group
= self
._scaling
_groups
[group_name
]
1663 scale_instance
= group
.create_instance(index
, is_default
)
1667 self
._log
.debug("Creating %u VNFs associated with NS id %s scaling group %s",
1668 len(self
.nsd_msg
.constituent_vnfd
), self
.id, self
)
1671 for vnf_index
, count
in group
.vnf_index_count_map
.items():
1672 const_vnfd_msg
= self
._get
_constituent
_vnfd
_msg
(vnf_index
)
1673 vnfd_msg
= self
._get
_vnfd
(const_vnfd_msg
.vnfd_id_ref
, config_xact
)
1675 cloud_account_name
, om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd_msg
.member_vnf_index
)
1676 if cloud_account_name
is None:
1677 cloud_account_name
= self
.cloud_account_name
1678 for _
in range(count
):
1679 vnfr
= yield from self
.create_vnf_record(vnfd_msg
, const_vnfd_msg
, cloud_account_name
, om_datacenter_name
, group_name
, index
)
1680 scale_instance
.add_vnfr(vnfr
)
1686 def instantiate_instance():
1687 self
._log
.debug("Creating %s VNFRS", scale_instance
)
1688 vnfrs
= yield from create_vnfs()
1689 yield from self
.publish()
1691 self
._log
.debug("Instantiating %s VNFRS for %s", len(vnfrs
), scale_instance
)
1692 scale_instance
.operational_status
= "vnf_init_phase"
1693 yield from self
.update_state()
1696 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
,
1697 group
, scale_instance
, vnfrs
)
1699 self
._log
.error("Pre scale out config for scale group {} ({}) failed".
1700 format(group
.name
, index
))
1701 scale_instance
.operational_status
= "failed"
1703 yield from self
.instantiate_vnfs(vnfrs
)
1705 except Exception as e
:
1706 self
._log
.exception("Failed to begin instantiatiation of vnfs for scale group {}: {}".
1707 format(group
.name
, e
))
1708 self
._log
.exception(e
)
1709 scale_instance
.operational_status
= "failed"
1711 yield from self
.update_state()
1713 yield from instantiate_instance()
1716 def delete_scale_group_instance(self
, group_name
, index
):
1717 group
= self
._scaling
_groups
[group_name
]
1718 scale_instance
= group
.get_instance(index
)
1719 if scale_instance
.is_default
:
1720 raise ScalingOperationError("Cannot terminate a default scaling group instance")
1722 scale_instance
.operational_status
= "terminate"
1723 yield from self
.update_state()
1726 def terminate_instance():
1727 self
._log
.debug("Terminating %s VNFRS" % scale_instance
)
1728 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_IN
,
1729 group
, scale_instance
)
1731 self
._log
.error("Pre scale in config for scale group {} ({}) failed".
1732 format(group
.name
, index
))
1734 # Going ahead with terminate, even if there is an error in pre-scale-in config
1735 # as this could be result of scale out failure and we need to cleanup this group
1736 yield from self
.terminate_vnfrs(scale_instance
.vnfrs
)
1737 group
.delete_instance(index
)
1739 scale_instance
.operational_status
= "vnf_terminate_phase"
1740 yield from self
.update_state()
1742 yield from terminate_instance()
1745 def _update_scale_group_instances_status(self
):
1747 def post_scale_out_task(group
, instance
):
1748 # Apply post scale out config once all VNFRs are active
1749 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_OUT
,
1751 instance
.operational_status
= "running"
1753 self
._log
.debug("Scale out for group {} and instance {} succeeded".
1754 format(group
.name
, instance
.instance_id
))
1756 self
._log
.error("Post scale out config for scale group {} ({}) failed".
1757 format(group
.name
, instance
.instance_id
))
1759 yield from self
.update_state()
1761 group_instances
= {group
: group
.instances
for group
in self
._scaling
_groups
.values()}
1762 for group
, instances
in group_instances
.items():
1763 self
._log
.debug("Updating %s instance status", group
)
1764 for instance
in instances
:
1765 instance_vnf_state_list
= [vnfr
.state
for vnfr
in instance
.vnfrs
]
1766 self
._log
.debug("Got vnfr instance states: %s", instance_vnf_state_list
)
1767 if instance
.operational_status
== "vnf_init_phase":
1768 if all([state
== VnfRecordState
.ACTIVE
for state
in instance_vnf_state_list
]):
1769 instance
.operational_status
= "running"
1771 # Create a task for post scale out to allow us to sleep before attempting
1772 # to configure newly created VM's
1773 self
._loop
.create_task(post_scale_out_task(group
, instance
))
1775 elif any([state
== VnfRecordState
.FAILED
for state
in instance_vnf_state_list
]):
1776 self
._log
.debug("Scale out for group {} and instance {} failed".
1777 format(group
.name
, instance
.instance_id
))
1778 instance
.operational_status
= "failed"
1780 elif instance
.operational_status
== "vnf_terminate_phase":
1781 if all([state
== VnfRecordState
.TERMINATED
for state
in instance_vnf_state_list
]):
1782 instance
.operational_status
= "terminated"
1783 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_IN
,
1786 self
._log
.debug("Scale in for group {} and instance {} succeeded".
1787 format(group
.name
, instance
.instance_id
))
1789 self
._log
.error("Post scale in config for scale group {} ({}) failed".
1790 format(group
.name
, instance
.instance_id
))
1792 def create_vnffgs(self
):
1793 """ This function creates VNFFGs for every VNFFG in the NSD
1794 associated with this NSR"""
1796 for vnffgd
in self
.nsd_msg
.vnffgd
:
1797 self
._log
.debug("Found vnffgd %s in nsr id %s", vnffgd
, self
.id)
1798 vnffgr
= VnffgRecord(self
._dts
,
1801 self
._nsm
._vnffgmgr
,
1805 self
._sdn
_account
_name
1807 self
._vnffgrs
[vnffgr
.id] = vnffgr
1809 def resolve_vld_ip_profile(self
, nsd_msg
, vld
):
1810 if not vld
.has_field('ip_profile_ref'):
1812 profile
= [ profile
for profile
in nsd_msg
.ip_profiles
if profile
.name
== vld
.ip_profile_ref
]
1813 return profile
[0] if profile
else None
1816 def _create_vls(self
, vld
, cloud_account
,om_datacenter
):
1817 """Create a VLR in the cloud account specified using the given VLD
1821 cloud_account : Cloud account name
1826 vlr
= yield from VirtualLinkRecord
.create_record(
1834 self
.resolve_vld_ip_profile(self
.nsd_msg
, vld
),
1836 restart_mode
=self
.restart_mode
)
1840 def _extract_cloud_accounts_for_vl(self
, vld
):
1842 Extracts the list of cloud accounts from the NS Config obj
1845 1. Cloud accounts based connection point (vnf_cloud_account_map)
1847 vld : VLD yang object
1852 cloud_account_list
= []
1854 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1855 # Handle case where cloud_account is None
1857 for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1858 if vnf
.cloud_account
is not None or vnf
.om_datacenter
is not None:
1859 vnf_cloud_map
[vnf
.member_vnf_index_ref
] = (vnf
.cloud_account
,vnf
.om_datacenter
)
1861 for vnfc
in vld
.vnfd_connection_point_ref
:
1862 cloud_account
= vnf_cloud_map
.get(
1863 vnfc
.member_vnf_index_ref
,
1864 (self
.cloud_account_name
,self
.om_datacenter_name
))
1866 cloud_account_list
.append(cloud_account
)
1868 if self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1869 for vld_map
in self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1870 if vld_map
.vld_id_ref
== vld
.id:
1871 for cloud_account
in vld_map
.cloud_accounts
:
1872 cloud_account_list
.extend((cloud_account
,None))
1873 for om_datacenter
in vld_map
.om_datacenters
:
1874 cloud_account_list
.extend((None,om_datacenter
))
1876 # If no config has been provided then fall-back to the default
1878 if not cloud_account_list
:
1879 cloud_account_list
= [(self
.cloud_account_name
,self
.om_datacenter_name
)]
1881 self
._log
.debug("VL {} cloud accounts: {}".
1882 format(vld
.name
, cloud_account_list
))
1883 return set(cloud_account_list
)
1886 def create_vls(self
):
1887 """ This function creates VLs for every VLD in the NSD
1888 associated with this NSR"""
1889 for vld
in self
.nsd_msg
.vld
:
1890 self
._log
.debug("Found vld %s in nsr id %s", vld
, self
.id)
1891 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1892 for cloud_account
,om_datacenter
in cloud_account_list
:
1893 vlr
= yield from self
._create
_vls
(vld
, cloud_account
,om_datacenter
)
1894 self
._vlrs
.append(vlr
)
1898 def create_vl_instance(self
, vld
):
1899 self
._log
.debug("Create VL for {}: {}".format(self
.id, vld
.as_dict()))
1900 # Check if the VL is already present
1902 for vl
in self
._vlrs
:
1903 if vl
.vld_msg
.id == vld
.id:
1904 self
._log
.debug("The VLD %s already in NSR %s as VLR %s with status %s",
1905 vld
.id, self
.id, vl
.id, vl
.state
)
1907 if vlr
.state
!= VlRecordState
.TERMINATED
:
1908 err_msg
= "VLR for VL %s in NSR %s already instantiated", \
1910 self
._log
.error(err_msg
)
1911 raise NsrVlUpdateError(err_msg
)
1915 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1916 for account
in cloud_account_list
:
1917 vlr
= yield from self
._create
_vls
(vld
, account
)
1918 self
._vlrs
.append(vlr
)
1920 vlr
.state
= VlRecordState
.INSTANTIATION_PENDING
1921 yield from self
.update_state()
1924 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1925 vlr
.state
= VlRecordState
.ACTIVE
1927 except Exception as e
:
1928 err_msg
= "Error instantiating VL for NSR {} and VLD {}: {}". \
1929 format(self
.id, vld
.id, e
)
1930 self
._log
.error(err_msg
)
1931 self
._log
.exception(e
)
1932 vlr
.state
= VlRecordState
.FAILED
1934 yield from self
.update_state()
1937 def delete_vl_instance(self
, vld
):
1938 for vlr
in self
._vlrs
:
1939 if vlr
.vld_msg
.id == vld
.id:
1940 self
._log
.debug("Found VLR %s for VLD %s in NSR %s",
1941 vlr
.id, vld
.id, self
.id)
1942 vlr
.state
= VlRecordState
.TERMINATE_PENDING
1943 yield from self
.update_state()
1946 yield from self
.nsm_plugin
.terminate_vl(vlr
)
1947 vlr
.state
= VlRecordState
.TERMINATED
1948 self
._vlrs
.remove(vlr
)
1950 except Exception as e
:
1951 err_msg
= "Error terminating VL for NSR {} and VLD {}: {}". \
1952 format(self
.id, vld
.id, e
)
1953 self
._log
.error(err_msg
)
1954 self
._log
.exception(e
)
1955 vlr
.state
= VlRecordState
.FAILED
1957 yield from self
.update_state()
1961 def create_vnfs(self
, config_xact
):
1963 This function creates VNFs for every VNF in the NSD
1964 associated with this NSR
1966 self
._log
.debug("Creating %u VNFs associated with this NS id %s",
1967 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
1969 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1970 if not const_vnfd
.start_by_default
:
1971 self
._log
.debug("start_by_default set to False in constituent VNF (%s). Skipping start.",
1972 const_vnfd
.member_vnf_index
)
1975 vnfd_msg
= self
._get
_vnfd
(const_vnfd
.vnfd_id_ref
, config_xact
)
1976 cloud_account_name
,om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd
.member_vnf_index
)
1977 if cloud_account_name
is None:
1978 cloud_account_name
= self
.cloud_account_name
1979 yield from self
.create_vnf_record(vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
)
1982 def get_placement_groups(self
, vnfd_msg
, const_vnfd
):
1983 placement_groups
= []
1984 for group
in self
.nsd_msg
.placement_groups
:
1985 for member_vnfd
in group
.member_vnfd
:
1986 if (member_vnfd
.vnfd_id_ref
== vnfd_msg
.id) and \
1987 (member_vnfd
.member_vnf_index_ref
== const_vnfd
.member_vnf_index
):
1988 group_info
= self
.resolve_placement_group_cloud_construct(group
)
1989 if group_info
is None:
1990 self
._log
.error("Could not resolve cloud-construct for placement group: %s", group
.name
)
1991 ### raise PlacementGroupError("Could not resolve cloud-construct for placement group: {}".format(group.name))
1993 self
._log
.info("Successfully resolved cloud construct for placement group: %s for VNF: %s (Member Index: %s)",
1996 const_vnfd
.member_vnf_index
)
1997 placement_groups
.append(group_info
)
1998 return placement_groups
2001 def create_vnf_record(self
, vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
, group_name
=None, group_instance_id
=None):
2002 # Fetch the VNFD associated with this VNF
2003 placement_groups
= self
.get_placement_groups(vnfd_msg
, const_vnfd
)
2004 self
._log
.info("Cloud Account for VNF %d is %s",const_vnfd
.member_vnf_index
,cloud_account_name
)
2005 self
._log
.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s",
2007 const_vnfd
.member_vnf_index
,
2008 [ group
.name
for group
in placement_groups
])
2009 vnfr
= yield from VirtualNetworkFunctionRecord
.create_record(self
._dts
,
2022 restart_mode
=self
.restart_mode
,
2024 if vnfr
.id in self
._vnfrs
:
2025 err
= "VNF with VNFR id %s already in vnf list" % (vnfr
.id,)
2026 raise NetworkServiceRecordError(err
)
2028 self
._vnfrs
[vnfr
.id] = vnfr
2029 self
._nsm
.vnfrs
[vnfr
.id] = vnfr
2031 yield from vnfr
.set_config_status(NsrYang
.ConfigStates
.INIT
)
2033 self
._log
.debug("Added VNFR %s to NSM VNFR list with id %s",
2039 def create_param_pools(self
):
2040 for param_pool
in self
.nsd_msg
.parameter_pool
:
2041 self
._log
.debug("Found parameter pool %s in nsr id %s", param_pool
, self
.id)
2043 start_value
= param_pool
.range.start_value
2044 end_value
= param_pool
.range.end_value
2045 if end_value
< start_value
:
2046 raise NetworkServiceRecordError(
2047 "Parameter pool %s has invalid range (start: {}, end: {})".format(
2048 start_value
, end_value
2052 self
._param
_pools
[param_pool
.name
] = config_value_pool
.ParameterValuePool(
2055 range(start_value
, end_value
)
2059 def fetch_vnfr(self
, vnfr_path
):
2060 """ Fetch VNFR record """
2062 self
._log
.debug("Fetching VNFR with key %s while instantiating %s",
2064 res_iter
= yield from self
._dts
.query_read(vnfr_path
, rwdts
.XactFlag
.MERGE
)
2066 for ent
in res_iter
:
2067 res
= yield from ent
2073 def instantiate_vnfs(self
, vnfrs
):
2075 This function instantiates VNFs for every VNF in this Network Service
2077 self
._log
.debug("Instantiating %u VNFs in NS %s", len(vnfrs
), self
.id)
2079 self
._log
.debug("Instantiating VNF: %s in NS %s", vnf
, self
.id)
2080 yield from self
.nsm_plugin
.instantiate_vnf(self
, vnf
)
2083 def instantiate_vnffgs(self
):
2085 This function instantiates VNFFGs for every VNFFG in this Network Service
2087 self
._log
.debug("Instantiating %u VNFFGs in NS %s",
2088 len(self
.nsd_msg
.vnffgd
), self
.id)
2089 for _
, vnfr
in self
.vnfrs
.items():
2090 while vnfr
.state
in [VnfRecordState
.INSTANTIATION_PENDING
, VnfRecordState
.INIT
]:
2091 self
._log
.debug("Received vnfr state for vnfr %s is %s; retrying",vnfr
.name
,vnfr
.state
)
2092 yield from asyncio
.sleep(2, loop
=self
._loop
)
2093 if vnfr
.state
== VnfRecordState
.ACTIVE
:
2094 self
._log
.debug("Received vnfr state for vnfr %s is %s ",vnfr
.name
,vnfr
.state
)
2097 self
._log
.debug("Received vnfr state for vnfr %s is %s; failing vnffg creation",vnfr
.name
,vnfr
.state
)
2098 self
._vnffgr
_state
= VnffgRecordState
.FAILED
2101 self
._log
.info("Waiting for 90 seconds for VMs to come up")
2102 yield from asyncio
.sleep(90, loop
=self
._loop
)
2103 self
._log
.info("Starting VNFFG orchestration")
2104 for vnffg
in self
._vnffgrs
.values():
2105 self
._log
.debug("Instantiating VNFFG: %s in NS %s", vnffg
, self
.id)
2106 yield from vnffg
.instantiate()
2109 def instantiate_scaling_instances(self
, config_xact
):
2110 """ Instantiate any default scaling instances in this Network Service """
2111 for group
in self
._scaling
_groups
.values():
2112 for i
in range(group
.min_instance_count
):
2113 self
._log
.debug("Instantiating %s default scaling instance %s", group
, i
)
2114 yield from self
.create_scale_group_instance(
2115 group
.name
, i
, config_xact
, is_default
=True
2118 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2119 if group_msg
.scaling_group_name_ref
!= group
.name
:
2122 for instance
in group_msg
.instance
:
2123 self
._log
.debug("Reloading %s scaling instance %s", group_msg
, instance
.id)
2124 yield from self
.create_scale_group_instance(
2125 group
.name
, instance
.id, config_xact
, is_default
=False
2128 def has_scaling_instances(self
):
2129 """ Return boolean indicating if the network service has default scaling groups """
2130 for group
in self
._scaling
_groups
.values():
2131 if group
.min_instance_count
> 0:
2134 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2135 if len(group_msg
.instance
) > 0:
2142 """ This function publishes this NSR """
2143 self
._nsr
_msg
= self
.create_msg()
2145 self
._log
.debug("Publishing the NSR with xpath %s and nsr %s",
2149 if self
._debug
_running
:
2150 self
._log
.debug("Publishing NSR in RUNNING state!")
2153 with self
._dts
.transaction() as xact
:
2154 yield from self
._nsm
.nsr_handler
.update(xact
, self
.nsr_xpath
, self
._nsr
_msg
)
2155 if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
:
2156 self
._debug
_running
= True
2159 def unpublish(self
, xact
):
2160 """ Unpublish this NSR object """
2161 self
._log
.debug("Unpublishing Network service id %s", self
.id)
2162 yield from self
._nsm
.nsr_handler
.delete(xact
, self
.nsr_xpath
)
2165 def nsr_xpath(self
):
2166 """ Returns the xpath associated with this NSR """
2168 "D,/nsr:ns-instance-opdata" +
2169 "/nsr:nsr[nsr:ns-instance-config-ref = '{}']"
2173 def xpath_from_nsr(nsr
):
2174 """ Returns the xpath associated with this NSR op data"""
2175 return (NetworkServiceRecord
.XPATH
+
2176 "[nsr:ns-instance-config-ref = '{}']").format(nsr
.id)
2179 def nsd_xpath(self
):
2180 """ Return NSD config xpath."""
2182 "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}']"
2183 ).format(self
.nsd_id
)
2186 def instantiate(self
, config_xact
):
2187 """"Instantiates a NetworkServiceRecord.
2189 This function instantiates a Network service
2190 which involves the following steps,
2192 * Instantiate every VL in NSD by sending create VLR request to DTS.
2193 * Instantiate every VNF in NSD by sending create VNF reuqest to DTS.
2194 * Publish the NSR details to DTS
2197 nsr: The NSR configuration request containing nsr-id and nsd
2198 config_xact: The configuration transaction which initiated the instatiation
2201 NetworkServiceRecordError if the NSR creation fails
2207 self
._log
.debug("Instantiating NS - %s xact - %s", self
, config_xact
)
2209 # Move the state to INIITALIZING
2210 self
.set_state(NetworkServiceRecordState
.INIT
)
2212 event_descr
= "Instantiation Request Received NSR Id:%s" % self
.id
2213 self
.record_event("instantiating", event_descr
)
2216 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
2219 # Update ref count if nsd present in catalog
2220 self
._nsm
.get_nsd_ref(self
.nsd_id
)
2222 except NetworkServiceDescriptorError
:
2223 # This could be an NSD not in the nsd-catalog
2226 # Merge any config and initial config primitive values
2227 self
.config_store
.merge_nsd_config(self
.nsd_msg
)
2228 self
._log
.debug("Merged NSD: {}".format(self
.nsd_msg
.as_dict()))
2230 event_descr
= "Fetched NSD with descriptor id %s" % self
.nsd_id
2231 self
.record_event("nsd-fetched", event_descr
)
2233 if self
._nsd
is None:
2234 msg
= "Failed to fetch NSD with nsd-id [%s] for nsr-id %s"
2235 self
._log
.debug(msg
, self
.nsd_id
, self
.id)
2236 raise NetworkServiceRecordError(self
)
2238 self
._log
.debug("Got nsd result %s", self
._nsd
)
2240 # Substitute any input parameters
2241 self
.substitute_input_parameters(self
._nsd
, self
._nsr
_cfg
_msg
)
2244 yield from self
.create(config_xact
)
2246 # Publish the NSR to DTS
2247 yield from self
.publish()
2250 def do_instantiate():
2252 Instantiate network service
2254 self
._log
.debug("Instantiating VLs nsr id [%s] nsd id [%s]",
2255 self
.id, self
.nsd_id
)
2257 # instantiate the VLs
2258 event_descr
= ("Instantiating %s external VLs for NSR id %s" %
2259 (len(self
.nsd_msg
.vld
), self
.id))
2260 self
.record_event("begin-external-vls-instantiation", event_descr
)
2262 self
.set_state(NetworkServiceRecordState
.VL_INIT_PHASE
)
2264 yield from self
.instantiate_vls()
2266 # Publish the NSR to DTS
2267 yield from self
.publish()
2269 event_descr
= ("Finished instantiating %s external VLs for NSR id %s" %
2270 (len(self
.nsd_msg
.vld
), self
.id))
2271 self
.record_event("end-external-vls-instantiation", event_descr
)
2273 self
.set_state(NetworkServiceRecordState
.VNF_INIT_PHASE
)
2275 self
._log
.debug("Instantiating VNFs ...... nsr[%s], nsd[%s]",
2276 self
.id, self
.nsd_id
)
2278 # instantiate the VNFs
2279 event_descr
= ("Instantiating %s VNFS for NSR id %s" %
2280 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2282 self
.record_event("begin-vnf-instantiation", event_descr
)
2284 yield from self
.instantiate_vnfs(self
._vnfrs
.values())
2286 self
._log
.debug(" Finished instantiating %d VNFs for NSR id %s",
2287 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
2289 event_descr
= ("Finished instantiating %s VNFs for NSR id %s" %
2290 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2291 self
.record_event("end-vnf-instantiation", event_descr
)
2293 if len(self
.vnffgrs
) > 0:
2294 #self.set_state(NetworkServiceRecordState.VNFFG_INIT_PHASE)
2295 event_descr
= ("Instantiating %s VNFFGS for NSR id %s" %
2296 (len(self
.nsd_msg
.vnffgd
), self
.id))
2298 self
.record_event("begin-vnffg-instantiation", event_descr
)
2300 yield from self
.instantiate_vnffgs()
2302 event_descr
= ("Finished instantiating %s VNFFGDs for NSR id %s" %
2303 (len(self
.nsd_msg
.vnffgd
), self
.id))
2304 self
.record_event("end-vnffg-instantiation", event_descr
)
2306 if self
.has_scaling_instances():
2307 event_descr
= ("Instantiating %s Scaling Groups for NSR id %s" %
2308 (len(self
._scaling
_groups
), self
.id))
2310 self
.record_event("begin-scaling-group-instantiation", event_descr
)
2311 yield from self
.instantiate_scaling_instances(config_xact
)
2312 self
.record_event("end-scaling-group-instantiation", event_descr
)
2314 # Give the plugin a chance to deploy the network service now that all
2315 # virtual links and vnfs are instantiated
2316 yield from self
.nsm_plugin
.deploy(self
._nsr
_msg
)
2318 self
._log
.debug("Publishing NSR...... nsr[%s], nsd[%s]",
2319 self
.id, self
.nsd_id
)
2321 # Publish the NSR to DTS
2322 yield from self
.publish()
2324 self
._log
.debug("Published NSR...... nsr[%s], nsd[%s]",
2325 self
.id, self
.nsd_id
)
2327 def on_instantiate_done(fut
):
2328 # If the do_instantiate fails, then publish NSR with failed result
2329 if fut
.exception() is not None:
2330 self
._log
.error("NSR instantiation failed for NSR id %s: %s", self
.id, str(fut
.exception()))
2331 self
._loop
.create_task(self
.instantiation_failed(failed_reason
=str(fut
.exception())))
2333 instantiate_task
= self
._loop
.create_task(do_instantiate())
2334 instantiate_task
.add_done_callback(on_instantiate_done
)
2337 def set_config_status(self
, status
, status_details
=None):
2338 if self
.config_status
!= status
:
2339 self
._log
.debug("Updating NSR {} status for {} to {}".
2340 format(self
.name
, self
.config_status
, status
))
2341 self
._config
_status
= status
2342 self
._config
_status
_details
= status_details
2344 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2345 self
.record_event("config-failed", "NS configuration failed",
2346 evt_details
=self
._config
_status
_details
)
2348 yield from self
.publish()
2351 def is_active(self
):
2352 """ This NS is active """
2353 self
.set_state(NetworkServiceRecordState
.RUNNING
)
2357 # Publish the NSR to DTS
2358 self
._log
.debug("Network service %s is active ", self
.id)
2359 self
._is
_active
= True
2361 event_descr
= "NSR in running state for NSR id %s" % self
.id
2362 self
.record_event("ns-running", event_descr
)
2364 yield from self
.publish()
2367 def instantiation_failed(self
, failed_reason
=None):
2368 """ The NS instantiation failed"""
2369 self
._log
.error("Network service id:%s, name:%s instantiation failed",
2371 self
.set_state(NetworkServiceRecordState
.FAILED
)
2373 event_descr
= "Instantiation of NS %s failed" % self
.id
2374 self
.record_event("ns-failed", event_descr
, evt_details
=failed_reason
)
2376 # Publish the NSR to DTS
2377 yield from self
.publish()
2380 def terminate_vnfrs(self
, vnfrs
):
2381 """ Terminate VNFRS in this network service """
2382 self
._log
.debug("Terminating VNFs in network service %s", self
.id)
2384 yield from self
.nsm_plugin
.terminate_vnf(vnfr
)
2387 def terminate(self
):
2388 """ Terminate a NetworkServiceRecord."""
2389 def terminate_vnffgrs():
2390 """ Terminate VNFFGRS in this network service """
2391 self
._log
.debug("Terminating VNFFGRs in network service %s", self
.id)
2392 for vnffgr
in self
.vnffgrs
.values():
2393 yield from vnffgr
.terminate()
2395 def terminate_vlrs():
2396 """ Terminate VLRs in this netork service """
2397 self
._log
.debug("Terminating VLs in network service %s", self
.id)
2398 for vlr
in self
.vlrs
:
2399 yield from self
.nsm_plugin
.terminate_vl(vlr
)
2400 vlr
.state
= VlRecordState
.TERMINATED
2402 self
._log
.debug("Terminating network service id %s", self
.id)
2404 # Move the state to TERMINATE
2405 self
.set_state(NetworkServiceRecordState
.TERMINATE
)
2406 event_descr
= "Terminate being processed for NS Id:%s" % self
.id
2407 self
.record_event("terminate", event_descr
)
2409 # Move the state to VNF_TERMINATE_PHASE
2410 self
._log
.debug("Terminating VNFFGs in NS ID: %s", self
.id)
2411 self
.set_state(NetworkServiceRecordState
.VNFFG_TERMINATE_PHASE
)
2412 event_descr
= "Terminating VNFFGS in NS Id:%s" % self
.id
2413 self
.record_event("terminating-vnffgss", event_descr
)
2414 yield from terminate_vnffgrs()
2416 # Move the state to VNF_TERMINATE_PHASE
2417 self
.set_state(NetworkServiceRecordState
.VNF_TERMINATE_PHASE
)
2418 event_descr
= "Terminating VNFS in NS Id:%s" % self
.id
2419 self
.record_event("terminating-vnfs", event_descr
)
2420 yield from self
.terminate_vnfrs(self
.vnfrs
.values())
2422 # Move the state to VL_TERMINATE_PHASE
2423 self
.set_state(NetworkServiceRecordState
.VL_TERMINATE_PHASE
)
2424 event_descr
= "Terminating VLs in NS Id:%s" % self
.id
2425 self
.record_event("terminating-vls", event_descr
)
2426 yield from terminate_vlrs()
2428 yield from self
.nsm_plugin
.terminate_ns(self
)
2430 # Move the state to TERMINATED
2431 self
.set_state(NetworkServiceRecordState
.TERMINATED
)
2432 event_descr
= "Terminated NS Id:%s" % self
.id
2433 self
.record_event("terminated", event_descr
)
2436 """"Enable a NetworkServiceRecord."""
2440 """"Disable a NetworkServiceRecord."""
2443 def map_config_status(self
):
2444 self
._log
.debug("Config status for ns {} is {}".
2445 format(self
.name
, self
._config
_status
))
2446 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURING
:
2447 return 'configuring'
2448 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2452 def vl_phase_completed(self
):
2453 """ Are VLs created in this NS?"""
2454 return self
._vl
_phase
_completed
2456 def vnf_phase_completed(self
):
2457 """ Are VLs created in this NS?"""
2458 return self
._vnf
_phase
_completed
2460 def create_msg(self
):
2461 """ The network serice record as a message """
2462 nsr_dict
= {"ns_instance_config_ref": self
.id}
2463 nsr
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr
.from_dict(nsr_dict
)
2464 #nsr.cloud_account = self.cloud_account_name
2465 nsr
.sdn_account
= self
._sdn
_account
_name
2466 nsr
.name_ref
= self
.name
2467 nsr
.nsd_ref
= self
.nsd_id
2468 nsr
.nsd_name_ref
= self
.nsd_msg
.name
2469 nsr
.operational_events
= self
._op
_status
.msg
2470 nsr
.operational_status
= self
._op
_status
.yang_str()
2471 nsr
.config_status
= self
.map_config_status()
2472 nsr
.config_status_details
= self
._config
_status
_details
2473 nsr
.create_time
= self
._create
_time
2475 for cfg_prim
in self
.nsd_msg
.service_primitive
:
2476 cfg_prim
= NsrYang
.YangData_Nsr_NsInstanceOpdata_Nsr_ServicePrimitive
.from_dict(
2478 nsr
.service_primitive
.append(cfg_prim
)
2480 for init_cfg
in self
.nsd_msg
.initial_config_primitive
:
2481 prim
= NsrYang
.NsrInitialConfigPrimitive
.from_dict(
2483 nsr
.initial_config_primitive
.append(prim
)
2485 if self
.vl_phase_completed():
2486 for vlr
in self
.vlrs
:
2487 nsr
.vlr
.append(vlr
.create_nsr_vlr_msg(self
.vnfrs
.values()))
2489 if self
.vnf_phase_completed():
2490 for vnfr_id
in self
.vnfrs
:
2491 nsr
.constituent_vnfr_ref
.append(self
.vnfrs
[vnfr_id
].const_vnfr_msg
)
2492 for vnffgr
in self
.vnffgrs
.values():
2493 nsr
.vnffgr
.append(vnffgr
.fetch_vnffgr())
2494 for scaling_group
in self
._scaling
_groups
.values():
2495 nsr
.scaling_group_record
.append(scaling_group
.create_record_msg())
2499 def all_vnfs_active(self
):
2500 """ Are all VNFS in this NS active? """
2501 for _
, vnfr
in self
.vnfrs
.items():
2502 if vnfr
.active
is not True:
2507 def update_state(self
):
2508 """ Re-evaluate this NS's state """
2509 curr_state
= self
._op
_status
.state
2511 if curr_state
== NetworkServiceRecordState
.TERMINATED
:
2512 self
._log
.debug("NS (%s) in terminated state, not updating state", self
.id)
2515 new_state
= NetworkServiceRecordState
.RUNNING
2516 self
._log
.info("Received update_state for nsr: %s, curr-state: %s",
2517 self
.id, curr_state
)
2519 # Check all the VNFRs are present
2520 for _
, vnfr
in self
.vnfrs
.items():
2521 if vnfr
.state
in [VnfRecordState
.ACTIVE
, VnfRecordState
.TERMINATED
]:
2523 elif vnfr
.state
== VnfRecordState
.FAILED
:
2524 if vnfr
._prev
_state
!= vnfr
.state
:
2525 event_descr
= "Instantiation of VNF %s failed" % vnfr
.id
2526 event_error_details
= vnfr
.state_failed_reason
2527 self
.record_event("vnf-failed", event_descr
, evt_details
=event_error_details
)
2528 vnfr
.set_state(VnfRecordState
.FAILED
)
2530 self
._log
.info("VNF state did not change, curr=%s, prev=%s",
2531 vnfr
.state
, vnfr
._prev
_state
)
2532 new_state
= NetworkServiceRecordState
.FAILED
2535 self
._log
.info("VNF %s in NSR %s is still not active; current state is: %s",
2536 vnfr
.id, self
.id, vnfr
.state
)
2537 new_state
= curr_state
2539 # If new state is RUNNING; check all VLs
2540 if new_state
== NetworkServiceRecordState
.RUNNING
:
2541 for vl
in self
.vlrs
:
2543 if vl
.state
in [VlRecordState
.ACTIVE
, VlRecordState
.TERMINATED
]:
2545 elif vl
.state
== VlRecordState
.FAILED
:
2546 if vl
.prev_state
!= vl
.state
:
2547 event_descr
= "Instantiation of VL %s failed" % vl
.id
2548 event_error_details
= vl
.state_failed_reason
2549 self
.record_event("vl-failed", event_descr
, evt_details
=event_error_details
)
2550 vl
.prev_state
= vl
.state
2552 self
._log
.debug("VL %s already in failed state")
2554 if vl
.state
in [VlRecordState
.INSTANTIATION_PENDING
, VlRecordState
.INIT
]:
2555 new_state
= NetworkServiceRecordState
.VL_INSTANTIATE
2558 if vl
.state
in [VlRecordState
.TERMINATE_PENDING
]:
2559 new_state
= NetworkServiceRecordState
.VL_TERMINATE
2562 # If new state is RUNNING; check VNFFGRs are also active
2563 if new_state
== NetworkServiceRecordState
.RUNNING
:
2564 for _
, vnffgr
in self
.vnffgrs
.items():
2565 self
._log
.info("Checking vnffgr state for nsr %s is: %s",
2566 self
.id, vnffgr
.state
)
2567 if vnffgr
.state
== VnffgRecordState
.ACTIVE
:
2569 elif vnffgr
.state
== VnffgRecordState
.FAILED
:
2570 event_descr
= "Instantiation of VNFFGR %s failed" % vnffgr
.id
2571 self
.record_event("vnffg-failed", event_descr
)
2572 new_state
= NetworkServiceRecordState
.FAILED
2575 self
._log
.info("VNFFGR %s in NSR %s is still not active; current state is: %s",
2576 vnffgr
.id, self
.id, vnffgr
.state
)
2577 new_state
= curr_state
2579 # Update all the scaling group instance operational status to
2580 # reflect the state of all VNFR within that instance
2581 yield from self
._update
_scale
_group
_instances
_status
()
2583 for _
, group
in self
._scaling
_groups
.items():
2584 if group
.state
== scale_group
.ScaleGroupState
.SCALING_OUT
:
2585 new_state
= NetworkServiceRecordState
.SCALING_OUT
2587 elif group
.state
== scale_group
.ScaleGroupState
.SCALING_IN
:
2588 new_state
= NetworkServiceRecordState
.SCALING_IN
2591 if new_state
!= curr_state
:
2592 self
._log
.debug("Changing state of Network service %s from %s to %s",
2593 self
.id, curr_state
, new_state
)
2594 if new_state
== NetworkServiceRecordState
.RUNNING
:
2595 yield from self
.is_active()
2596 elif new_state
== NetworkServiceRecordState
.FAILED
:
2597 # If the NS is already active and we entered scaling_in, scaling_out,
2598 # do not mark the NS as failing if scaling operation failed.
2599 if curr_state
in [NetworkServiceRecordState
.SCALING_OUT
,
2600 NetworkServiceRecordState
.SCALING_IN
] and self
._is
_active
:
2601 new_state
= NetworkServiceRecordState
.RUNNING
2602 self
.set_state(new_state
)
2604 yield from self
.instantiation_failed()
2606 self
.set_state(new_state
)
2608 yield from self
.publish()
2611 class InputParameterSubstitution(object):
2613 This class is responsible for substituting input parameters into an NSD.
2616 def __init__(self
, log
):
2617 """Create an instance of InputParameterSubstitution
2620 log - a logger for this object to use
2625 def __call__(self
, nsd
, nsr_config
):
2626 """Substitutes input parameters from the NSR config into the NSD
2628 This call modifies the provided NSD with the input parameters that are
2629 contained in the NSR config.
2632 nsd - a GI NSD object
2633 nsr_config - a GI NSR config object
2636 if nsd
is None or nsr_config
is None:
2639 # Create a lookup of the xpath elements that this descriptor allows
2641 optional_input_parameters
= set()
2642 for input_parameter
in nsd
.input_parameter_xpath
:
2643 optional_input_parameters
.add(input_parameter
.xpath
)
2645 # Apply the input parameters to the descriptor
2646 if nsr_config
.input_parameter
:
2647 for param
in nsr_config
.input_parameter
:
2648 if param
.xpath
not in optional_input_parameters
:
2649 msg
= "tried to set an invalid input parameter ({})"
2650 self
.log
.error(msg
.format(param
.xpath
))
2654 "input-parameter:{} = {}".format(
2661 xpath
.setxattr(nsd
, param
.xpath
, param
.value
)
2663 except Exception as e
:
2664 self
.log
.exception(e
)
2667 class NetworkServiceDescriptor(object):
2669 Network service descriptor class
2672 def __init__(self
, dts
, log
, loop
, nsd
, nsm
):
2684 """ Returns nsd id """
2689 """ Returns name of nsd """
2690 return self
._nsd
.name
2693 def ref_count(self
):
2694 """ Returns reference count"""
2695 return self
._ref
_count
2698 """ Returns whether nsd is in use or not """
2699 return True if self
.ref_count
> 0 else False
2702 """ Take a reference on this object """
2703 self
._ref
_count
+= 1
2706 """ Release reference on this object """
2707 if self
.ref_count
< 1:
2708 msg
= ("Unref on a NSD object - nsd id %s, ref_count = %s" %
2709 (self
.id, self
.ref_count
))
2710 self
._log
.critical(msg
)
2711 raise NetworkServiceDescriptorError(msg
)
2712 self
._ref
_count
-= 1
2716 """ Return the message associated with this NetworkServiceDescriptor"""
2720 def path_for_id(nsd_id
):
2721 """ Return path for the passed nsd_id"""
2722 return "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}'".format(nsd_id
)
2725 """ Return the message associated with this NetworkServiceDescriptor"""
2726 return NetworkServiceDescriptor
.path_for_id(self
.id)
2728 def update(self
, nsd
):
2729 """ Update the NSD descriptor """
2733 class NsdDtsHandler(object):
2734 """ The network service descriptor DTS handler """
2735 XPATH
= "C,/nsd:nsd-catalog/nsd:nsd"
2737 def __init__(self
, dts
, log
, loop
, nsm
):
2747 """ Return registration handle """
2752 """ Register for Nsd create/update/delete/read requests from dts """
2754 def on_apply(dts
, acg
, xact
, action
, scratch
):
2755 """Apply the configuration"""
2756 is_recovery
= xact
.xact
is None and action
== rwdts
.AppconfAction
.INSTALL
2757 self
._log
.debug("Got nsd apply cfg (xact:%s) (action:%s)",
2759 # Create/Update an NSD record
2760 for cfg
in self
._regh
.get_xact_elements(xact
):
2761 # Only interested in those NSD cfgs whose ID was received in prepare callback
2762 if cfg
.id in scratch
.get('nsds', []) or is_recovery
:
2763 self
._nsm
.update_nsd(cfg
)
2765 scratch
.pop('nsds', None)
2767 return RwTypes
.RwStatus
.SUCCESS
2770 def delete_nsd_libs(nsd_id
):
2771 """ Remove any files uploaded with NSD and stored under $RIFT_ARTIFACTS/libs/<id> """
2773 rift_artifacts_dir
= os
.environ
['RIFT_ARTIFACTS']
2774 nsd_dir
= os
.path
.join(rift_artifacts_dir
, 'launchpad/libs', nsd_id
)
2776 if os
.path
.exists (nsd_dir
):
2777 shutil
.rmtree(nsd_dir
, ignore_errors
=True)
2778 except Exception as e
:
2779 self
._log
.error("Exception in cleaning up NSD libs {}: {}".
2781 self
._log
.excpetion(e
)
2784 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2785 """ Prepare callback from DTS for NSD config """
2787 self
._log
.info("Got nsd prepare - config received nsd id %s, msg %s",
2790 fref
= ProtobufC
.FieldReference
.alloc()
2791 fref
.goto_whole_message(msg
.to_pbcm())
2793 if fref
.is_field_deleted():
2794 # Delete an NSD record
2795 self
._log
.debug("Deleting NSD with id %s", msg
.id)
2796 if self
._nsm
.nsd_in_use(msg
.id):
2797 self
._log
.debug("Cannot delete NSD in use - %s", msg
.id)
2798 err
= "Cannot delete an NSD in use - %s" % msg
.id
2799 raise NetworkServiceDescriptorRefCountExists(err
)
2801 yield from delete_nsd_libs(msg
.id)
2802 self
._nsm
.delete_nsd(msg
.id)
2804 # Add this NSD to scratch to create/update in apply callback
2805 nsds
= scratch
.setdefault('nsds', [])
2807 # acg._scratch['nsds'].append(msg.id)
2809 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2812 "Registering for NSD config using xpath: %s",
2813 NsdDtsHandler
.XPATH
,
2816 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2817 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2818 # Need a list in scratch to store NSDs to create/update later
2819 # acg._scratch['nsds'] = list()
2820 self
._regh
= acg
.register(
2821 xpath
=NsdDtsHandler
.XPATH
,
2822 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
2823 on_prepare
=on_prepare
)
2826 class VnfdDtsHandler(object):
2827 """ DTS handler for VNFD config changes """
2828 XPATH
= "C,/vnfd:vnfd-catalog/vnfd:vnfd"
2830 def __init__(self
, dts
, log
, loop
, nsm
):
2839 """ DTS registration handle """
2844 """ Register for VNFD configuration"""
2847 def on_apply(dts
, acg
, xact
, action
, scratch
):
2848 """Apply the configuration"""
2849 self
._log
.debug("Got NSM VNFD apply (xact: %s) (action: %s)(scr: %s)",
2850 xact
, action
, scratch
)
2852 # Create/Update a VNFD record
2853 for cfg
in self
._regh
.get_xact_elements(xact
):
2854 # Only interested in those VNFD cfgs whose ID was received in prepare callback
2855 if cfg
.id in scratch
.get('vnfds', []):
2856 self
._nsm
.update_vnfd(cfg
)
2858 for cfg
in self
._regh
.elements
:
2859 if cfg
.id in scratch
.get('deleted_vnfds', []):
2860 yield from self
._nsm
.delete_vnfd(cfg
.id)
2862 scratch
.pop('vnfds', None)
2863 scratch
.pop('deleted_vnfds', None)
2866 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2867 """ on prepare callback """
2868 self
._log
.debug("Got on prepare for VNFD (path: %s) (action: %s) (msg: %s)",
2869 ks_path
.to_xpath(RwNsmYang
.get_schema()), xact_info
.query_action
, msg
)
2871 fref
= ProtobufC
.FieldReference
.alloc()
2872 fref
.goto_whole_message(msg
.to_pbcm())
2874 # Handle deletes in prepare_callback, but adds/updates in apply_callback
2875 if fref
.is_field_deleted():
2876 self
._log
.debug("Adding msg to deleted field")
2877 deleted_vnfds
= scratch
.setdefault('deleted_vnfds', [])
2878 deleted_vnfds
.append(msg
.id)
2880 # Add this VNFD to scratch to create/update in apply callback
2881 vnfds
= scratch
.setdefault('vnfds', [])
2882 vnfds
.append(msg
.id)
2884 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2887 "Registering for VNFD config using xpath: %s",
2888 VnfdDtsHandler
.XPATH
,
2890 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2891 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2892 # Need a list in scratch to store VNFDs to create/update later
2893 # acg._scratch['vnfds'] = list()
2894 # acg._scratch['deleted_vnfds'] = list()
2895 self
._regh
= acg
.register(
2896 xpath
=VnfdDtsHandler
.XPATH
,
2897 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY
,
2898 on_prepare
=on_prepare
)
2900 class NsrRpcDtsHandler(object):
2901 """ The network service instantiation RPC DTS handler """
2902 EXEC_NSR_CONF_XPATH
= "I,/nsr:start-network-service"
2903 EXEC_NSR_CONF_O_XPATH
= "O,/nsr:start-network-service"
2904 NETCONF_IP_ADDRESS
= "127.0.0.1"
2906 RESTCONF_PORT
= 8888
2907 NETCONF_USER
= "admin"
2908 NETCONF_PW
= "admin"
2909 REST_BASE_V2_URL
= 'https://{}:{}/v2/api/'.format("127.0.0.1",8888)
2911 def __init__(self
, dts
, log
, loop
, nsm
):
2918 self
._ns
_regh
= None
2920 self
._manager
= None
2921 self
._nsr
_config
_url
= NsrRpcDtsHandler
.REST_BASE_V2_URL
+ 'config/ns-instance-config'
2923 self
._model
= RwYang
.Model
.create_libncx()
2924 self
._model
.load_schema_ypbc(RwNsrYang
.get_schema())
2928 """ Return the NS manager instance """
2932 def wrap_netconf_config_xml(xml
):
2933 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
2937 def _connect(self
, timeout_secs
=240):
2939 start_time
= time
.time()
2940 while (time
.time() - start_time
) < timeout_secs
:
2943 self
._log
.debug("Attemping NsmTasklet netconf connection.")
2945 manager
= yield from ncclient
.asyncio_manager
.asyncio_connect(
2947 host
=NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
,
2948 port
=NsrRpcDtsHandler
.NETCONF_PORT
,
2949 username
=NsrRpcDtsHandler
.NETCONF_USER
,
2950 password
=NsrRpcDtsHandler
.NETCONF_PW
,
2952 look_for_keys
=False,
2953 hostkey_verify
=False,
2958 except ncclient
.transport
.errors
.SSHError
as e
:
2959 self
._log
.warning("Netconf connection to launchpad %s failed: %s",
2960 NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
, str(e
))
2962 yield from asyncio
.sleep(5, loop
=self
._loop
)
2964 raise NsrInstantiationFailed("Failed to connect to Launchpad within %s seconds" %
2967 def _apply_ns_instance_config(self
,payload_dict
):
2968 #self._log.debug("At apply NS instance config with payload %s",payload_dict)
2969 req_hdr
= {'accept':'application/vnd.yang.data+json','content-type':'application/vnd.yang.data+json'}
2970 response
=requests
.post(self
._nsr
_config
_url
, headers
=req_hdr
, auth
=('admin', 'admin'),data
=payload_dict
,verify
=False)
2975 """ Register for NS monitoring read from dts """
2977 def on_ns_config_prepare(xact_info
, action
, ks_path
, msg
):
2978 """ prepare callback from dts start-network-service"""
2979 assert action
== rwdts
.QueryAction
.RPC
2981 rpc_op
= NsrYang
.YangOutput_Nsr_StartNetworkService
.from_dict({
2982 "nsr_id":str(uuid
.uuid4())
2985 if not ('name' in rpc_ip
and 'nsd_ref' in rpc_ip
and ('cloud_account' in rpc_ip
or 'om_datacenter' in rpc_ip
)):
2986 self
._log
.error("Mandatory parameters name or nsd_ref or cloud account not found in start-network-service {}".format(rpc_ip
))
2989 self
._log
.debug("start-network-service RPC input: {}".format(rpc_ip
))
2992 # Add used value to the pool
2993 self
._log
.debug("RPC output: {}".format(rpc_op
))
2995 nsd_copy
= self
.nsm
.get_nsd(rpc_ip
.nsd_ref
)
2997 #if not self._manager:
2998 # self._manager = yield from self._connect()
3000 self
._log
.debug("Configuring ns-instance-config with name %s nsd-ref: %s",
3001 rpc_ip
.name
, rpc_ip
.nsd_ref
)
3003 ns_instance_config_dict
= {"id":rpc_op
.nsr_id
, "admin_status":"ENABLED"}
3004 ns_instance_config_copy_dict
= {k
:v
for k
, v
in rpc_ip
.as_dict().items()
3005 if k
in RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr().fields
}
3006 ns_instance_config_dict
.update(ns_instance_config_copy_dict
)
3008 ns_instance_config
= RwNsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.from_dict(ns_instance_config_dict
)
3009 ns_instance_config
.nsd
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_Nsd()
3010 ns_instance_config
.nsd
.from_dict(nsd_copy
.msg
.as_dict())
3012 payload_dict
= ns_instance_config
.to_json(self
._model
)
3013 #xml = ns_instance_config.to_xml_v2(self._model)
3014 #netconf_xml = self.wrap_netconf_config_xml(xml)
3016 #self._log.debug("Sending configure ns-instance-config xml to %s: %s",
3017 # netconf_xml, NsrRpcDtsHandler.NETCONF_IP_ADDRESS)
3018 self
._log
.debug("Sending configure ns-instance-config json to %s: %s",
3019 self
._nsr
_config
_url
,ns_instance_config
)
3021 #response = yield from self._manager.edit_config(
3023 # config=netconf_xml,
3025 response
= yield from self
._loop
.run_in_executor(
3027 self
._apply
_ns
_instance
_config
,
3030 response
.raise_for_status()
3031 self
._log
.debug("Received edit config response: %s", response
.json())
3033 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
,
3034 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
,
3036 except Exception as e
:
3037 self
._log
.error("Exception processing the "
3038 "start-network-service: {}".format(e
))
3039 self
._log
.exception(e
)
3040 xact_info
.respond_xpath(rwdts
.XactRspCode
.NACK
,
3041 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
)
3044 hdl_ns
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_ns_config_prepare
,)
3046 with self
._dts
.group_create() as group
:
3047 self
._ns
_regh
= group
.register(xpath
=NsrRpcDtsHandler
.EXEC_NSR_CONF_XPATH
,
3049 flags
=rwdts
.Flag
.PUBLISHER
,
3053 class NsrDtsHandler(object):
3054 """ The network service DTS handler """
3055 NSR_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr"
3056 SCALE_INSTANCE_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr/nsr:scaling-group/nsr:instance"
3057 KEY_PAIR_XPATH
= "C,/nsr:key-pair"
3059 def __init__(self
, dts
, log
, loop
, nsm
):
3065 self
._nsr
_regh
= None
3066 self
._scale
_regh
= None
3067 self
._key
_pair
_regh
= None
3071 """ Return the NS manager instance """
3076 """ Register for Nsr create/update/delete/read requests from dts """
3078 def nsr_id_from_keyspec(ks
):
3079 nsr_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr
.schema().keyspec_to_entry(ks
)
3080 nsr_id
= nsr_path_entry
.key00
.id
3083 def group_name_from_keyspec(ks
):
3084 group_path_entry
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
.schema().keyspec_to_entry(ks
)
3085 group_name
= group_path_entry
.key00
.scaling_group_name_ref
3088 def is_instance_in_reg_elements(nsr_id
, group_name
, instance_id
):
3089 """ Return boolean indicating if scaling group instance was already commited previously.
3091 By looking at the existing elements in this registration handle (elements not part
3092 of this current xact), we can tell if the instance was configured previously without
3093 keeping any application state.
3095 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3096 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3097 elem_group_name
= group_name_from_keyspec(keyspec
)
3099 if elem_nsr_id
!= nsr_id
or group_name
!= elem_group_name
:
3102 if instance_cfg
.id == instance_id
:
3107 def get_scale_group_instance_delta(nsr_id
, group_name
, xact
):
3108 delta
= {"added": [], "deleted": []}
3109 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3110 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3111 if elem_nsr_id
!= nsr_id
:
3114 elem_group_name
= group_name_from_keyspec(keyspec
)
3115 if elem_group_name
!= group_name
:
3118 delta
["added"].append(instance_cfg
.id)
3120 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(include_keyspec
=True):
3121 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3122 if elem_nsr_id
!= nsr_id
:
3125 elem_group_name
= group_name_from_keyspec(keyspec
)
3126 if elem_group_name
!= group_name
:
3129 if instance_cfg
.id in delta
["added"]:
3130 delta
["added"].remove(instance_cfg
.id)
3132 delta
["deleted"].append(instance_cfg
.id)
3137 def update_nsr_nsd(nsr_id
, xact
, scratch
):
3140 def get_nsr_vl_delta(nsr_id
, xact
, scratch
):
3141 delta
= {"added": [], "deleted": []}
3142 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3143 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3144 if elem_nsr_id
!= nsr_id
:
3147 if 'vld' in instance_cfg
.nsd
:
3148 for vld
in instance_cfg
.nsd
.vld
:
3149 delta
["added"].append(vld
)
3151 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3152 self
._log
.debug("NSR update: %s", instance_cfg
)
3153 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3154 if elem_nsr_id
!= nsr_id
:
3157 if 'vld' in instance_cfg
.nsd
:
3158 for vld
in instance_cfg
.nsd
.vld
:
3159 if vld
in delta
["added"]:
3160 delta
["added"].remove(vld
)
3162 delta
["deleted"].append(vld
)
3166 vl_delta
= yield from get_nsr_vl_delta(nsr_id
, xact
, scratch
)
3167 self
._log
.debug("Got NSR:%s VL instance delta: %s", nsr_id
, vl_delta
)
3169 for vld
in vl_delta
["added"]:
3170 yield from self
._nsm
.nsr_instantiate_vl(nsr_id
, vld
)
3172 for vld
in vl_delta
["deleted"]:
3173 yield from self
._nsm
.nsr_terminate_vl(nsr_id
, vld
)
3175 def get_add_delete_update_cfgs(dts_member_reg
, xact
, key_name
, scratch
):
3176 # Unfortunately, it is currently difficult to figure out what has exactly
3177 # changed in this xact without Pbdelta support (RIFT-4916)
3178 # As a workaround, we can fetch the pre and post xact elements and
3179 # perform a comparison to figure out adds/deletes/updates
3180 xact_cfgs
= list(dts_member_reg
.get_xact_elements(xact
))
3181 curr_cfgs
= list(dts_member_reg
.elements
)
3183 xact_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in xact_cfgs
}
3184 curr_key_map
= {getattr(cfg
, key_name
): cfg
for cfg
in curr_cfgs
}
3187 added_keys
= set(xact_key_map
) - set(curr_key_map
)
3188 added_cfgs
= [xact_key_map
[key
] for key
in added_keys
]
3191 deleted_keys
= set(curr_key_map
) - set(xact_key_map
)
3192 deleted_cfgs
= [curr_key_map
[key
] for key
in deleted_keys
]
3195 updated_keys
= set(curr_key_map
) & set(xact_key_map
)
3196 updated_cfgs
= [xact_key_map
[key
] for key
in updated_keys
3197 if xact_key_map
[key
] != curr_key_map
[key
]]
3199 return added_cfgs
, deleted_cfgs
, updated_cfgs
3201 def get_nsr_key_pairs(dts_member_reg
, xact
):
3203 for instance_cfg
, keyspec
in dts_member_reg
.get_xact_elements(xact
, include_keyspec
=True):
3204 self
._log
.debug("Key pair received is {} KS: {}".format(instance_cfg
, keyspec
))
3205 xpath
= keyspec
.to_xpath(RwNsrYang
.get_schema())
3206 key_pairs
[instance_cfg
.name
] = instance_cfg
3209 def on_apply(dts
, acg
, xact
, action
, scratch
):
3210 """Apply the configuration"""
3211 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3212 xact
, action
, scratch
)
3214 def handle_create_nsr(msg
, key_pairs
=None, restart_mode
=False):
3215 # Handle create nsr requests """
3216 # Do some validations
3217 if not msg
.has_field("nsd"):
3218 err
= "NSD not provided"
3219 self
._log
.error(err
)
3220 raise NetworkServiceRecordError(err
)
3222 self
._log
.debug("Creating NetworkServiceRecord %s from nsr config %s",
3223 msg
.id, msg
.as_dict())
3224 nsr
= self
.nsm
.create_nsr(msg
, key_pairs
=key_pairs
, restart_mode
=restart_mode
)
3227 def handle_delete_nsr(msg
):
3229 def delete_instantiation(ns_id
):
3230 """ Delete instantiation """
3231 with self
._dts
.transaction() as xact
:
3232 yield from self
._nsm
.terminate_ns(ns_id
, xact
)
3234 # Handle delete NSR requests
3235 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3236 # Terminate the NSR instance
3237 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3239 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3240 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3241 nsr
.record_event("terminate-rcvd", event_descr
)
3243 self
._loop
.create_task(delete_instantiation(msg
.id))
3246 def begin_instantiation(nsr
):
3247 # Begin instantiation
3248 self
._log
.info("Beginning NS instantiation: %s", nsr
.id)
3249 yield from self
._nsm
.instantiate_ns(nsr
.id, xact
)
3251 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3252 xact
, action
, scratch
)
3254 if action
== rwdts
.AppconfAction
.INSTALL
and xact
.id is None:
3256 for element
in self
._key
_pair
_regh
.elements
:
3257 key_pairs
.append(element
)
3258 for element
in self
._nsr
_regh
.elements
:
3259 nsr
= handle_create_nsr(element
, key_pairs
, restart_mode
=True)
3260 self
._loop
.create_task(begin_instantiation(nsr
))
3263 (added_msgs
, deleted_msgs
, updated_msgs
) = get_add_delete_update_cfgs(self
._nsr
_regh
,
3267 self
._log
.debug("Added: %s, Deleted: %s, Updated: %s", added_msgs
,
3268 deleted_msgs
, updated_msgs
)
3270 for msg
in added_msgs
:
3271 if msg
.id not in self
._nsm
.nsrs
:
3272 self
._log
.info("Create NSR received in on_apply to instantiate NS:%s", msg
.id)
3273 key_pairs
= get_nsr_key_pairs(self
._key
_pair
_regh
, xact
)
3274 nsr
= handle_create_nsr(msg
,key_pairs
)
3275 self
._loop
.create_task(begin_instantiation(nsr
))
3277 for msg
in deleted_msgs
:
3278 self
._log
.info("Delete NSR received in on_apply to terminate NS:%s", msg
.id)
3280 handle_delete_nsr(msg
)
3282 self
._log
.exception("Failed to terminate NS:%s", msg
.id)
3284 for msg
in updated_msgs
:
3285 self
._log
.info("Update NSR received in on_apply: %s", msg
)
3287 self
._nsm
.nsr_update_cfg(msg
.id, msg
)
3290 self
._loop
.create_task(update_nsr_nsd(msg
.id, xact
, scratch
))
3292 for group
in msg
.scaling_group
:
3293 instance_delta
= get_scale_group_instance_delta(msg
.id, group
.scaling_group_name_ref
, xact
)
3294 self
._log
.debug("Got NSR:%s scale group instance delta: %s", msg
.id, instance_delta
)
3296 for instance_id
in instance_delta
["added"]:
3297 self
._nsm
.scale_nsr_out(msg
.id, group
.scaling_group_name_ref
, instance_id
, xact
)
3299 for instance_id
in instance_delta
["deleted"]:
3300 self
._nsm
.scale_nsr_in(msg
.id, group
.scaling_group_name_ref
, instance_id
)
3303 return RwTypes
.RwStatus
.SUCCESS
3306 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
3307 """ Prepare calllback from DTS for NSR """
3309 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3310 action
= xact_info
.query_action
3312 "Got Nsr prepare callback (xact: %s) (action: %s) (info: %s), %s:%s)",
3313 xact
, action
, xact_info
, xpath
, msg
3317 def delete_instantiation(ns_id
):
3318 """ Delete instantiation """
3319 yield from self
._nsm
.terminate_ns(ns_id
, None)
3321 def handle_delete_nsr():
3322 """ Handle delete NSR requests """
3323 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3324 # Terminate the NSR instance
3325 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3327 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3328 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3329 nsr
.record_event("terminate-rcvd", event_descr
)
3331 self
._loop
.create_task(delete_instantiation(msg
.id))
3333 fref
= ProtobufC
.FieldReference
.alloc()
3334 fref
.goto_whole_message(msg
.to_pbcm())
3336 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
, rwdts
.QueryAction
.DELETE
]:
3337 # if this is an NSR create
3338 if action
!= rwdts
.QueryAction
.DELETE
and msg
.id not in self
._nsm
.nsrs
:
3339 # Ensure the Cloud account/datacenter has been specified
3340 if not msg
.has_field("cloud_account") and not msg
.has_field("om_datacenter"):
3341 raise NsrInstantiationFailed("Cloud account or datacenter not specified in NSR")
3343 # Check if nsd is specified
3344 if not msg
.has_field("nsd"):
3345 raise NsrInstantiationFailed("NSD not specified in NSR")
3348 nsr
= self
._nsm
.nsrs
[msg
.id]
3350 if msg
.has_field("nsd"):
3351 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3352 raise NsrVlUpdateError("Unable to update VL when NSR not in running state")
3353 if 'vld' not in msg
.nsd
or len(msg
.nsd
.vld
) == 0:
3354 raise NsrVlUpdateError("NS config NSD should have atleast 1 VLD defined")
3356 if msg
.has_field("scaling_group"):
3357 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3358 raise ScalingOperationError("Unable to perform scaling action when NS is not in running state")
3360 if len(msg
.scaling_group
) > 1:
3361 raise ScalingOperationError("Only a single scaling group can be configured at a time")
3363 for group_msg
in msg
.scaling_group
:
3364 num_new_group_instances
= len(group_msg
.instance
)
3365 if num_new_group_instances
> 1:
3366 raise ScalingOperationError("Only a single scaling instance can be modified at a time")
3368 elif num_new_group_instances
== 1:
3369 scale_group
= nsr
.scaling_groups
[group_msg
.scaling_group_name_ref
]
3370 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
]:
3371 if len(scale_group
.instances
) == scale_group
.max_instance_count
:
3372 raise ScalingOperationError("Max instances for %s reached" % scale_group
)
3374 acg
.handle
.prepare_complete_ok(xact_info
.handle
)
3377 self
._log
.debug("Registering for NSR config using xpath: %s",
3378 NsrDtsHandler
.NSR_XPATH
)
3380 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
3381 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
3382 self
._nsr
_regh
= acg
.register(xpath
=NsrDtsHandler
.NSR_XPATH
,
3383 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3384 on_prepare
=on_prepare
)
3386 self
._scale
_regh
= acg
.register(
3387 xpath
=NsrDtsHandler
.SCALE_INSTANCE_XPATH
,
3388 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY| rwdts
.Flag
.CACHE
,
3391 self
._key
_pair
_regh
= acg
.register(
3392 xpath
=NsrDtsHandler
.KEY_PAIR_XPATH
,
3393 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3397 class NsrOpDataDtsHandler(object):
3398 """ The network service op data DTS handler """
3399 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
3401 def __init__(self
, dts
, log
, loop
, nsm
):
3410 """ Return the registration handle"""
3415 """ Return the NS manager instance """
3420 """ Register for Nsr op data publisher registration"""
3421 self
._log
.debug("Registering Nsr op data path %s as publisher",
3422 NsrOpDataDtsHandler
.XPATH
)
3424 hdl
= rift
.tasklets
.DTS
.RegistrationHandler()
3425 handlers
= rift
.tasklets
.Group
.Handler()
3426 with self
._dts
.group_create(handler
=handlers
) as group
:
3427 self
._regh
= group
.register(xpath
=NsrOpDataDtsHandler
.XPATH
,
3429 flags
=rwdts
.Flag
.PUBLISHER | rwdts
.Flag
.NO_PREP_READ | rwdts
.Flag
.DATASTORE
)
3432 def create(self
, path
, msg
):
3434 Create an NS record in DTS with the path and message
3436 self
._log
.debug("Creating NSR %s:%s", path
, msg
)
3437 self
.regh
.create_element(path
, msg
)
3438 self
._log
.debug("Created NSR, %s:%s", path
, msg
)
3441 def update(self
, path
, msg
, flags
=rwdts
.XactFlag
.REPLACE
):
3443 Update an NS record in DTS with the path and message
3445 self
._log
.debug("Updating NSR, %s:%s regh = %s", path
, msg
, self
.regh
)
3446 self
.regh
.update_element(path
, msg
, flags
)
3447 self
._log
.debug("Updated NSR, %s:%s", path
, msg
)
3450 def delete(self
, path
):
3452 Update an NS record in DTS with the path and message
3454 self
._log
.debug("Deleting NSR path:%s", path
)
3455 self
.regh
.delete_element(path
)
3456 self
._log
.debug("Deleted NSR path:%s", path
)
3459 class VnfrDtsHandler(object):
3460 """ The virtual network service DTS handler """
3461 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
3463 def __init__(self
, dts
, log
, loop
, nsm
):
3473 """ Return registration handle """
3478 """ Return the NS manager instance """
3483 """ Register for vnfr create/update/delete/ advises from dts """
3485 def on_commit(xact_info
):
3486 """ The transaction has been committed """
3487 self
._log
.debug("Got vnfr commit (xact_info: %s)", xact_info
)
3488 return rwdts
.MemberRspCode
.ACTION_OK
3491 def on_prepare(xact_info
, action
, ks_path
, msg
):
3492 """ prepare callback from dts """
3493 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3495 "Got vnfr on_prepare cb (xact_info: %s, action: %s): %s:%s",
3496 xact_info
, action
, ks_path
, msg
3499 schema
= VnfrYang
.YangData_Vnfr_VnfrCatalog_Vnfr
.schema()
3500 path_entry
= schema
.keyspec_to_entry(ks_path
)
3501 if path_entry
.key00
.id not in self
._nsm
._vnfrs
:
3502 self
._log
.error("%s request for non existent record path %s",
3504 xact_info
.respond_xpath(rwdts
.XactRspCode
.NA
, xpath
)
3508 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3509 if action
== rwdts
.QueryAction
.CREATE
or action
== rwdts
.QueryAction
.UPDATE
:
3510 yield from self
._nsm
.update_vnfr(msg
)
3511 elif action
== rwdts
.QueryAction
.DELETE
:
3512 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3513 self
._nsm
.delete_vnfr(path_entry
.key00
.id)
3515 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
, xpath
)
3517 self
._log
.debug("Registering for VNFR using xpath: %s",
3518 VnfrDtsHandler
.XPATH
,)
3520 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_commit
=on_commit
,
3521 on_prepare
=on_prepare
,)
3522 with self
._dts
.group_create() as group
:
3523 self
._regh
= group
.register(xpath
=VnfrDtsHandler
.XPATH
,
3525 flags
=(rwdts
.Flag
.SUBSCRIBER
),)
3528 class NsdRefCountDtsHandler(object):
3529 """ The NSD Ref Count DTS handler """
3530 XPATH
= "D,/nsr:ns-instance-opdata/rw-nsr:nsd-ref-count"
3532 def __init__(self
, dts
, log
, loop
, nsm
):
3542 """ Return registration handle """
3547 """ Return the NS manager instance """
3552 """ Register for NSD ref count read from dts """
3555 def on_prepare(xact_info
, action
, ks_path
, msg
):
3556 """ prepare callback from dts """
3557 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3559 if action
== rwdts
.QueryAction
.READ
:
3560 schema
= RwNsrYang
.YangData_Nsr_NsInstanceOpdata_NsdRefCount
.schema()
3561 path_entry
= schema
.keyspec_to_entry(ks_path
)
3562 nsd_list
= yield from self
._nsm
.get_nsd_refcount(path_entry
.key00
.nsd_id_ref
)
3563 for xpath
, msg
in nsd_list
:
3564 xact_info
.respond_xpath(rsp_code
=rwdts
.XactRspCode
.MORE
,
3567 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
3569 raise NetworkServiceRecordError("Not supported operation %s" % action
)
3571 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_prepare
,)
3572 with self
._dts
.group_create() as group
:
3573 self
._regh
= group
.register(xpath
=NsdRefCountDtsHandler
.XPATH
,
3575 flags
=rwdts
.Flag
.PUBLISHER
,)
3578 class NsManager(object):
3579 """ The Network Service Manager class"""
3580 def __init__(self
, dts
, log
, loop
,
3581 nsr_handler
, vnfr_handler
, vlr_handler
, ro_plugin_selector
,
3582 vnffgmgr
, vnfd_pub_handler
, cloud_account_handler
):
3586 self
._nsr
_handler
= nsr_handler
3587 self
._vnfr
_pub
_handler
= vnfr_handler
3588 self
._vlr
_pub
_handler
= vlr_handler
3589 self
._vnffgmgr
= vnffgmgr
3590 self
._vnfd
_pub
_handler
= vnfd_pub_handler
3591 self
._cloud
_account
_handler
= cloud_account_handler
3593 self
._ro
_plugin
_selector
= ro_plugin_selector
3594 self
._ncclient
= rift
.mano
.ncclient
.NcClient(
3606 self
.cfgmgr_obj
= conman
.ROConfigManager(log
, loop
, dts
, self
)
3608 # TODO: All these handlers should move to tasklet level.
3609 # Passing self is often an indication of bad design
3610 self
._nsd
_dts
_handler
= NsdDtsHandler(dts
, log
, loop
, self
)
3611 self
._vnfd
_dts
_handler
= VnfdDtsHandler(dts
, log
, loop
, self
)
3612 self
._dts
_handlers
= [self
._nsd
_dts
_handler
,
3613 VnfrDtsHandler(dts
, log
, loop
, self
),
3614 NsdRefCountDtsHandler(dts
, log
, loop
, self
),
3615 NsrDtsHandler(dts
, log
, loop
, self
),
3616 ScalingRpcHandler(log
, dts
, loop
, self
.scale_rpc_callback
),
3617 NsrRpcDtsHandler(dts
,log
,loop
,self
),
3618 self
._vnfd
_dts
_handler
,
3639 def nsr_handler(self
):
3640 """" NSR handler """
3641 return self
._nsr
_handler
3645 """" So Obj handler """
3650 """ NSRs in this NSM"""
3655 """ NSDs in this NSM"""
3660 """ VNFDs in this NSM"""
3665 """ VNFRs in this NSM"""
3669 def nsr_pub_handler(self
):
3670 """ NSR publication handler """
3671 return self
._nsr
_handler
3674 def vnfr_pub_handler(self
):
3675 """ VNFR publication handler """
3676 return self
._vnfr
_pub
_handler
3679 def vlr_pub_handler(self
):
3680 """ VLR publication handler """
3681 return self
._vlr
_pub
_handler
3684 def vnfd_pub_handler(self
):
3685 return self
._vnfd
_pub
_handler
3689 """ Register all static DTS handlers """
3690 for dts_handle
in self
._dts
_handlers
:
3691 yield from dts_handle
.register()
3694 def get_ns_by_nsr_id(self
, nsr_id
):
3695 """ get NSR by nsr id """
3696 if nsr_id
not in self
._nsrs
:
3697 raise NetworkServiceRecordError("NSR id %s not found" % nsr_id
)
3699 return self
._nsrs
[nsr_id
]
3701 def scale_nsr_out(self
, nsr_id
, scale_group_name
, instance_id
, config_xact
):
3702 self
.log
.debug("Scale out NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3707 nsr
= self
._nsrs
[nsr_id
]
3708 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3709 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3711 self
._loop
.create_task(nsr
.create_scale_group_instance(scale_group_name
, instance_id
, config_xact
))
3713 def scale_nsr_in(self
, nsr_id
, scale_group_name
, instance_id
):
3714 self
.log
.debug("Scale in NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3719 nsr
= self
._nsrs
[nsr_id
]
3720 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3721 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3723 self
._loop
.create_task(nsr
.delete_scale_group_instance(scale_group_name
, instance_id
))
3725 def scale_rpc_callback(self
, xact
, msg
, action
):
3726 """Callback handler for RPC calls
3728 xact : Transaction Handler
3730 action : Scaling Action
3732 ScalingGroupInstance
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup_Instance
3733 ScalingGroup
= NsrYang
.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
3735 xpath
= ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').format(
3737 instance
= ScalingGroupInstance
.from_dict({"id": msg
.instance_id
})
3740 def get_nsr_scaling_group():
3741 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3743 for result
in results
:
3744 res
= yield from result
3745 nsr_config
= res
.result
3747 for scaling_group
in nsr_config
.scaling_group
:
3748 if scaling_group
.scaling_group_name_ref
== msg
.scaling_group_name_ref
:
3751 scaling_group
= nsr_config
.scaling_group
.add()
3752 scaling_group
.scaling_group_name_ref
= msg
.scaling_group_name_ref
3754 return (nsr_config
, scaling_group
)
3757 def update_config(nsr_config
):
3758 xml
= self
._ncclient
.convert_to_xml(RwNsrYang
, nsr_config
)
3759 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
3760 yield from self
._ncclient
.connect()
3761 yield from self
._ncclient
.manager
.edit_config(target
="running", config
=xml
, default_operation
="replace")
3765 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3766 scaling_group
.instance
.append(instance
)
3767 yield from update_config(nsr_config
)
3771 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3772 scaling_group
.instance
.remove(instance
)
3773 yield from update_config(nsr_config
)
3775 if action
== ScalingRpcHandler
.ACTION
.SCALE_OUT
:
3776 self
._loop
.create_task(scale_out())
3778 self
._loop
.create_task(scale_in())
3780 # Opdata based calls, disabled for now!
3781 # if action == ScalingRpcHandler.ACTION.SCALE_OUT:
3782 # self.scale_nsr_out(
3784 # msg.scaling_group_name_ref,
3788 # self.scale_nsr_in(
3790 # msg.scaling_group_name_ref,
3793 def nsr_update_cfg(self
, nsr_id
, msg
):
3794 nsr
= self
._nsrs
[nsr_id
]
3795 nsr
.nsr_cfg_msg
= msg
3797 def nsr_instantiate_vl(self
, nsr_id
, vld
):
3798 self
.log
.debug("NSR {} create VL {}".format(nsr_id
, vld
))
3799 nsr
= self
._nsrs
[nsr_id
]
3800 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3801 raise NsrVlUpdateError("Cannot perform VL instantiate if NSR is not in running state")
3803 # Not calling in a separate task as this is called from a separate task
3804 yield from nsr
.create_vl_instance(vld
)
3806 def nsr_terminate_vl(self
, nsr_id
, vld
):
3807 self
.log
.debug("NSR {} delete VL {}".format(nsr_id
, vld
.id))
3808 nsr
= self
._nsrs
[nsr_id
]
3809 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3810 raise NsrVlUpdateError("Cannot perform VL terminate if NSR is not in running state")
3812 # Not calling in a separate task as this is called from a separate task
3813 yield from nsr
.delete_vl_instance(vld
)
3815 def create_nsr(self
, nsr_msg
, key_pairs
=None,restart_mode
=False):
3816 """ Create an NSR instance """
3817 if nsr_msg
.id in self
._nsrs
:
3818 msg
= "NSR id %s already exists" % nsr_msg
.id
3819 self
._log
.error(msg
)
3820 raise NetworkServiceRecordError(msg
)
3822 self
._log
.info("Create NetworkServiceRecord nsr id %s from nsd_id %s",
3826 nsm_plugin
= self
._ro
_plugin
_selector
.ro_plugin
3827 sdn_account_name
= self
._cloud
_account
_handler
.get_cloud_account_sdn_name(nsr_msg
.cloud_account
)
3829 nsr
= NetworkServiceRecord(self
._dts
,
3837 restart_mode
=restart_mode
3839 self
._nsrs
[nsr_msg
.id] = nsr
3840 nsm_plugin
.create_nsr(nsr_msg
, nsr_msg
.nsd
, key_pairs
)
3844 def delete_nsr(self
, nsr_id
):
3846 Delete NSR with the passed nsr id
3848 del self
._nsrs
[nsr_id
]
3851 def instantiate_ns(self
, nsr_id
, config_xact
):
3852 """ Instantiate an NS instance """
3853 self
._log
.debug("Instantiating Network service id %s", nsr_id
)
3854 if nsr_id
not in self
._nsrs
:
3855 err
= "NSR id %s not found " % nsr_id
3856 self
._log
.error(err
)
3857 raise NetworkServiceRecordError(err
)
3859 nsr
= self
._nsrs
[nsr_id
]
3860 yield from nsr
.nsm_plugin
.instantiate_ns(nsr
, config_xact
)
3863 def update_vnfr(self
, vnfr
):
3864 """Create/Update an VNFR """
3866 vnfr_state
= self
._vnfrs
[vnfr
.id].state
3867 self
._log
.debug("Updating VNFR with state %s: vnfr %s", vnfr_state
, vnfr
)
3869 yield from self
._vnfrs
[vnfr
.id].update_state(vnfr
)
3870 nsr
= self
.find_nsr_for_vnfr(vnfr
.id)
3871 yield from nsr
.update_state()
3873 def find_nsr_for_vnfr(self
, vnfr_id
):
3874 """ Find the NSR which )has the passed vnfr id"""
3875 for nsr
in list(self
.nsrs
.values()):
3876 for vnfr
in list(nsr
.vnfrs
.values()):
3877 if vnfr
.id == vnfr_id
:
3881 def delete_vnfr(self
, vnfr_id
):
3882 """ Delete VNFR with the passed id"""
3883 del self
._vnfrs
[vnfr_id
]
3885 def get_nsd_ref(self
, nsd_id
):
3886 """ Get network service descriptor for the passed nsd_id
3888 nsd
= self
.get_nsd(nsd_id
)
3893 def get_nsr_config(self
, nsd_id
):
3894 xpath
= "C,/nsr:ns-instance-config"
3895 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3897 for result
in results
:
3898 entry
= yield from result
3899 ns_instance_config
= entry
.result
3901 for nsr
in ns_instance_config
.nsr
:
3902 if nsr
.nsd
.id == nsd_id
:
3908 def nsd_unref_by_nsr_id(self
, nsr_id
):
3909 """ Unref the network service descriptor based on NSR id """
3910 self
._log
.debug("NSR Unref called for Nsr Id:%s", nsr_id
)
3911 if nsr_id
in self
._nsrs
:
3912 nsr
= self
._nsrs
[nsr_id
]
3915 nsd
= self
.get_nsd(nsr
.nsd_id
)
3916 self
._log
.debug("Releasing ref on NSD %s held by NSR %s - Curr %d",
3917 nsd
.id, nsr
.id, nsd
.ref_count
)
3919 except NetworkServiceDescriptorError
:
3920 # We store a copy of NSD in NSR and the NSD in nsd-catalog
3925 self
._log
.error("Cannot find NSR with id %s", nsr_id
)
3926 raise NetworkServiceDescriptorUnrefError("No NSR with id" % nsr_id
)
3929 def nsd_unref(self
, nsd_id
):
3930 """ Unref the network service descriptor associated with the id """
3931 nsd
= self
.get_nsd(nsd_id
)
3934 def get_nsd(self
, nsd_id
):
3935 """ Get network service descriptor for the passed nsd_id"""
3936 if nsd_id
not in self
._nsds
:
3937 self
._log
.error("Cannot find NSD id:%s", nsd_id
)
3938 raise NetworkServiceDescriptorError("Cannot find NSD id:%s", nsd_id
)
3940 return self
._nsds
[nsd_id
]
3942 def create_nsd(self
, nsd_msg
):
3943 """ Create a network service descriptor """
3944 self
._log
.debug("Create network service descriptor - %s", nsd_msg
)
3945 if nsd_msg
.id in self
._nsds
:
3946 self
._log
.error("Cannot create NSD %s -NSD ID already exists", nsd_msg
)
3947 raise NetworkServiceDescriptorError("NSD already exists-%s", nsd_msg
.id)
3949 nsd
= NetworkServiceDescriptor(
3956 self
._nsds
[nsd_msg
.id] = nsd
3960 def update_nsd(self
, nsd
):
3961 """ update the Network service descriptor """
3962 self
._log
.debug("Update network service descriptor - %s", nsd
)
3963 if nsd
.id not in self
._nsds
:
3964 self
._log
.debug("No NSD found - creating NSD id = %s", nsd
.id)
3965 self
.create_nsd(nsd
)
3967 self
._log
.debug("Updating NSD id = %s, nsd = %s", nsd
.id, nsd
)
3968 self
._nsds
[nsd
.id].update(nsd
)
3970 def delete_nsd(self
, nsd_id
):
3971 """ Delete the Network service descriptor with the passed id """
3972 self
._log
.debug("Deleting the network service descriptor - %s", nsd_id
)
3973 if nsd_id
not in self
._nsds
:
3974 self
._log
.debug("Delete NSD failed - cannot find nsd-id %s", nsd_id
)
3975 raise NetworkServiceDescriptorNotFound("Cannot find %s", nsd_id
)
3977 if nsd_id
not in self
._nsds
:
3978 self
._log
.debug("Cannot delete NSD id %s reference exists %s",
3980 self
._nsds
[nsd_id
].ref_count
)
3981 raise NetworkServiceDescriptorRefCountExists(
3982 "Cannot delete :%s, ref_count:%s",
3984 self
._nsds
[nsd_id
].ref_count
)
3986 del self
._nsds
[nsd_id
]
3988 def get_vnfd_config(self
, xact
):
3989 vnfd_dts_reg
= self
._vnfd
_dts
_handler
.regh
3990 for cfg
in vnfd_dts_reg
.get_xact_elements(xact
):
3991 if cfg
.id not in self
._vnfds
:
3992 self
.create_vnfd(cfg
)
3994 def get_vnfd(self
, vnfd_id
, xact
):
3995 """ Get virtual network function descriptor for the passed vnfd_id"""
3996 if vnfd_id
not in self
._vnfds
:
3997 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
3998 self
.get_vnfd_config(xact
)
4000 if vnfd_id
not in self
._vnfds
:
4001 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4002 raise VnfDescriptorError("Cannot find VNFD id:%s", vnfd_id
)
4004 return self
._vnfds
[vnfd_id
]
4006 def create_vnfd(self
, vnfd
):
4007 """ Create a virtual network function descriptor """
4008 self
._log
.debug("Create virtual network function descriptor - %s", vnfd
)
4009 if vnfd
.id in self
._vnfds
:
4010 self
._log
.error("Cannot create VNFD %s -VNFD ID already exists", vnfd
)
4011 raise VnfDescriptorError("VNFD already exists-%s", vnfd
.id)
4013 self
._vnfds
[vnfd
.id] = vnfd
4014 return self
._vnfds
[vnfd
.id]
4016 def update_vnfd(self
, vnfd
):
4017 """ Update the virtual network function descriptor """
4018 self
._log
.debug("Update virtual network function descriptor- %s", vnfd
)
4020 # Hack to remove duplicates from leaf-lists - to be fixed by RIFT-6511
4021 for ivld
in vnfd
.internal_vld
:
4022 ivld
.internal_connection_point_ref
= list(set(ivld
.internal_connection_point_ref
))
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
)