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('ProjectNsdYang', '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 (
49 ProjectNsdYang
as NsdYang
,
61 import rift
.mano
.ncclient
62 import rift
.mano
.config_data
.config
63 import rift
.mano
.dts
as mano_dts
64 from rift
.mano
.utils
.project
import (
67 get_add_delete_update_cfgs
,
71 from . import rwnsm_conman
as conman
73 from . import publisher
75 from . import config_value_pool
76 from . import rwvnffgmgr
77 from . import scale_group
80 class NetworkServiceRecordState(Enum
):
81 """ Network Service Record State """
85 VNFFG_INIT_PHASE
= 104
91 VL_TERMINATE_PHASE
= 111
92 VNF_TERMINATE_PHASE
= 112
93 VNFFG_TERMINATE_PHASE
= 113
100 class NetworkServiceRecordError(Exception):
101 """ Network Service Record Error """
105 class NetworkServiceDescriptorError(Exception):
106 """ Network Service Descriptor Error """
110 class VirtualNetworkFunctionRecordError(Exception):
111 """ Virtual Network Function Record Error """
115 class NetworkServiceDescriptorNotFound(Exception):
116 """ Cannot find Network Service Descriptor"""
120 class NetworkServiceDescriptorRefCountExists(Exception):
121 """ Network Service Descriptor reference count exists """
125 class NetworkServiceDescriptorUnrefError(Exception):
126 """ Failed to unref a network service descriptor """
130 class NsrInstantiationFailed(Exception):
131 """ Failed to instantiate network service """
135 class VnfInstantiationFailed(Exception):
136 """ Failed to instantiate virtual network function"""
140 class VnffgInstantiationFailed(Exception):
141 """ Failed to instantiate virtual network function"""
145 class VnfDescriptorError(Exception):
146 """Failed to instantiate virtual network function"""
150 class ScalingOperationError(Exception):
154 class ScaleGroupMissingError(Exception):
158 class PlacementGroupError(Exception):
162 class NsrNsdUpdateError(Exception):
166 class NsrVlUpdateError(NsrNsdUpdateError
):
170 class VlRecordState(Enum
):
171 """ VL Record State """
173 INSTANTIATION_PENDING
= 102
175 TERMINATE_PENDING
= 104
180 class VnffgRecordState(Enum
):
181 """ VNFFG Record State """
183 INSTANTIATION_PENDING
= 102
185 TERMINATE_PENDING
= 104
190 class VnffgRecord(object):
191 """ Vnffg Records class"""
194 def __init__(self
, dts
, log
, loop
, vnffgmgr
, nsr
, nsr_name
, vnffgd_msg
, sdn_account_name
):
199 self
._vnffgmgr
= vnffgmgr
201 self
._nsr
_name
= nsr_name
202 self
._vnffgd
_msg
= vnffgd_msg
203 if sdn_account_name
is None:
204 self
._sdn
_account
_name
= ''
206 self
._sdn
_account
_name
= sdn_account_name
208 self
._vnffgr
_id
= str(uuid
.uuid4())
209 self
._vnffgr
_rsp
_id
= list()
210 self
._vnffgr
_state
= VnffgRecordState
.INIT
215 return self
._vnffgr
_id
219 """ state of this VNF """
220 return self
._vnffgr
_state
222 def fetch_vnffgr(self
):
224 Get VNFFGR message to be published
227 if self
._vnffgr
_state
== VnffgRecordState
.INIT
:
228 vnffgr_dict
= {"id": self
._vnffgr
_id
,
229 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
230 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
231 "sdn_account": self
._sdn
_account
_name
,
232 "operational_status": 'init',
234 vnffgr
= NsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
235 elif self
._vnffgr
_state
== VnffgRecordState
.TERMINATED
:
236 vnffgr_dict
= {"id": self
._vnffgr
_id
,
237 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
238 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
239 "sdn_account": self
._sdn
_account
_name
,
240 "operational_status": 'terminated',
242 vnffgr
= NsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
245 vnffgr
= self
._vnffgmgr
.fetch_vnffgr(self
._vnffgr
_id
)
247 self
._log
.exception("Fetching VNFFGR for VNFFG with id %s failed", self
._vnffgr
_id
)
248 self
._vnffgr
_state
= VnffgRecordState
.FAILED
249 vnffgr_dict
= {"id": self
._vnffgr
_id
,
250 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
251 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
252 "sdn_account": self
._sdn
_account
_name
,
253 "operational_status": 'failed',
255 vnffgr
= NsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
260 def vnffgr_create_msg(self
):
261 """ Virtual Link Record message for Creating VLR in VNS """
262 vnffgr_dict
= {"id": self
._vnffgr
_id
,
263 "vnffgd_id_ref": self
._vnffgd
_msg
.id,
264 "vnffgd_name_ref": self
._vnffgd
_msg
.name
,
265 "sdn_account": self
._sdn
_account
_name
,
267 vnffgr
= NsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr_Vnffgr
.from_dict(vnffgr_dict
)
268 for rsp
in self
._vnffgd
_msg
.rsp
:
269 vnffgr_rsp
= vnffgr
.rsp
.add()
270 vnffgr_rsp
.id = str(uuid
.uuid4())
271 vnffgr_rsp
.name
= self
._nsr
.name
+ '.' + rsp
.name
272 self
._vnffgr
_rsp
_id
.append(vnffgr_rsp
.id)
273 vnffgr_rsp
.vnffgd_rsp_id_ref
= rsp
.id
274 vnffgr_rsp
.vnffgd_rsp_name_ref
= rsp
.name
275 for rsp_cp_ref
in rsp
.vnfd_connection_point_ref
:
276 vnfd
= [vnfr
.vnfd
for vnfr
in self
._nsr
.vnfrs
.values() if vnfr
.vnfd
.id == rsp_cp_ref
.vnfd_id_ref
]
277 self
._log
.debug("VNFD message during VNFFG instantiation is %s",vnfd
)
278 if len(vnfd
) > 0 and vnfd
[0].has_field('service_function_type'):
279 self
._log
.debug("Service Function Type for VNFD ID %s is %s",
280 rsp_cp_ref
.vnfd_id_ref
, vnfd
[0].service_function_type
)
282 self
._log
.error("Service Function Type not available for VNFD ID %s; Skipping in chain",
283 rsp_cp_ref
.vnfd_id_ref
)
286 vnfr_cp_ref
= vnffgr_rsp
.vnfr_connection_point_ref
.add()
287 vnfr_cp_ref
.member_vnf_index_ref
= rsp_cp_ref
.member_vnf_index_ref
288 vnfr_cp_ref
.hop_number
= rsp_cp_ref
.order
289 vnfr_cp_ref
.vnfd_id_ref
=rsp_cp_ref
.vnfd_id_ref
290 vnfr_cp_ref
.service_function_type
= vnfd
[0].service_function_type
291 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
292 if (nsr_vnfr
.vnfd
.id == vnfr_cp_ref
.vnfd_id_ref
and
293 nsr_vnfr
.member_vnf_index
== vnfr_cp_ref
.member_vnf_index_ref
):
294 vnfr_cp_ref
.vnfr_id_ref
= nsr_vnfr
.id
295 vnfr_cp_ref
.vnfr_name_ref
= nsr_vnfr
.name
296 vnfr_cp_ref
.vnfr_connection_point_ref
= rsp_cp_ref
.vnfd_connection_point_ref
298 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
299 self
._log
.debug(" Received VNFR is %s", vnfr
)
300 while vnfr
.operational_status
!= 'running':
301 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
302 if vnfr
.operational_status
== 'failed':
303 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
304 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" %
306 yield from asyncio
.sleep(2, loop
=self
._loop
)
307 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
308 self
._log
.debug("Received VNFR is %s", vnfr
)
310 vnfr_cp_ref
.connection_point_params
.mgmt_address
= vnfr
.mgmt_interface
.ip_address
311 for cp
in vnfr
.connection_point
:
312 if cp
.name
== vnfr_cp_ref
.vnfr_connection_point_ref
:
313 vnfr_cp_ref
.connection_point_params
.port_id
= cp
.connection_point_id
314 vnfr_cp_ref
.connection_point_params
.name
= self
._nsr
.name
+ '.' + cp
.name
315 for vdu
in vnfr
.vdur
:
316 for ext_intf
in vdu
.external_interface
:
317 if ext_intf
.name
== vnfr_cp_ref
.vnfr_connection_point_ref
:
318 vnfr_cp_ref
.connection_point_params
.vm_id
= vdu
.vim_id
319 self
._log
.debug("VIM ID for CP %s in VNFR %s is %s",cp
.name
,nsr_vnfr
.id,
320 vnfr_cp_ref
.connection_point_params
.vm_id
)
323 vnfr_cp_ref
.connection_point_params
.address
= cp
.ip_address
324 vnfr_cp_ref
.connection_point_params
.port
= VnffgRecord
.SFF_DP_PORT
326 for vnffgd_classifier
in self
._vnffgd
_msg
.classifier
:
327 _rsp
= [rsp
for rsp
in vnffgr
.rsp
if rsp
.vnffgd_rsp_id_ref
== vnffgd_classifier
.rsp_id_ref
]
329 rsp_id_ref
= _rsp
[0].id
330 rsp_name
= _rsp
[0].name
332 self
._log
.error("RSP with ID %s not found during classifier creation for classifier id %s",
333 vnffgd_classifier
.rsp_id_ref
,vnffgd_classifier
.id)
335 vnffgr_classifier
= vnffgr
.classifier
.add()
336 vnffgr_classifier
.id = vnffgd_classifier
.id
337 vnffgr_classifier
.name
= self
._nsr
.name
+ '.' + vnffgd_classifier
.name
338 _rsp
[0].classifier_name
= vnffgr_classifier
.name
339 vnffgr_classifier
.rsp_id_ref
= rsp_id_ref
340 vnffgr_classifier
.rsp_name
= rsp_name
341 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
342 if (nsr_vnfr
.vnfd
.id == vnffgd_classifier
.vnfd_id_ref
and
343 nsr_vnfr
.member_vnf_index
== vnffgd_classifier
.member_vnf_index_ref
):
344 vnffgr_classifier
.vnfr_id_ref
= nsr_vnfr
.id
345 vnffgr_classifier
.vnfr_name_ref
= nsr_vnfr
.name
346 vnffgr_classifier
.vnfr_connection_point_ref
= vnffgd_classifier
.vnfd_connection_point_ref
348 if nsr_vnfr
.vnfd
.service_function_chain
== 'CLASSIFIER':
349 vnffgr_classifier
.sff_name
= nsr_vnfr
.name
351 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
352 self
._log
.debug(" Received VNFR is %s", vnfr
)
353 while vnfr
.operational_status
!= 'running':
354 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
355 if vnfr
.operational_status
== 'failed':
356 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
357 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" %
359 yield from asyncio
.sleep(2, loop
=self
._loop
)
360 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
361 self
._log
.debug("Received VNFR is %s", vnfr
)
363 for cp
in vnfr
.connection_point
:
364 if cp
.name
== vnffgr_classifier
.vnfr_connection_point_ref
:
365 vnffgr_classifier
.port_id
= cp
.connection_point_id
366 vnffgr_classifier
.ip_address
= cp
.ip_address
367 for vdu
in vnfr
.vdur
:
368 for ext_intf
in vdu
.external_interface
:
369 if ext_intf
.name
== vnffgr_classifier
.vnfr_connection_point_ref
:
370 vnffgr_classifier
.vm_id
= vdu
.vim_id
371 self
._log
.debug("VIM ID for CP %s in VNFR %s is %s",
373 vnfr_cp_ref
.connection_point_params
.vm_id
)
376 self
._log
.info("VNFFGR msg to be sent is %s", vnffgr
)
380 def vnffgr_nsr_sff_list(self
):
381 """ SFF List for VNFR """
383 sf_list
= [nsr_vnfr
.name
for nsr_vnfr
in self
._nsr
.vnfrs
.values() if nsr_vnfr
.vnfd
.service_function_chain
== 'SF']
385 for nsr_vnfr
in self
._nsr
.vnfrs
.values():
386 if (nsr_vnfr
.vnfd
.service_function_chain
== 'CLASSIFIER' or nsr_vnfr
.vnfd
.service_function_chain
== 'SFF'):
387 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
388 self
._log
.debug(" Received VNFR is %s", vnfr
)
389 while vnfr
.operational_status
!= 'running':
390 self
._log
.info("Received vnf op status is %s; retrying",vnfr
.operational_status
)
391 if vnfr
.operational_status
== 'failed':
392 self
._log
.error("Fetching VNFR for %s failed", vnfr
.id)
393 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self
.id, vnfr
.id))
394 yield from asyncio
.sleep(2, loop
=self
._loop
)
395 vnfr
= yield from self
._nsr
.fetch_vnfr(nsr_vnfr
.xpath
)
396 self
._log
.debug("Received VNFR is %s", vnfr
)
398 sff
= RwsdnYang
.VNFFGSff()
399 sff_list
[nsr_vnfr
.vnfd
.id] = sff
400 sff
.name
= nsr_vnfr
.name
401 sff
.function_type
= nsr_vnfr
.vnfd
.service_function_chain
403 sff
.mgmt_address
= vnfr
.mgmt_interface
.ip_address
404 sff
.mgmt_port
= VnffgRecord
.SFF_MGMT_PORT
405 for cp
in vnfr
.connection_point
:
406 sff_dp
= sff
.dp_endpoints
.add()
407 sff_dp
.name
= self
._nsr
.name
+ '.' + cp
.name
408 sff_dp
.address
= cp
.ip_address
409 sff_dp
.port
= VnffgRecord
.SFF_DP_PORT
410 if nsr_vnfr
.vnfd
.service_function_chain
== 'SFF':
411 for sf_name
in sf_list
:
412 _sf
= sff
.vnfr_list
.add()
413 _sf
.vnfr_name
= sf_name
418 def instantiate(self
):
419 """ Instantiate this VNFFG """
421 self
._log
.info("Instaniating VNFFGR with vnffgd %s",
425 vnffgr_request
= yield from self
.vnffgr_create_msg()
426 vnffg_sff_list
= yield from self
.vnffgr_nsr_sff_list()
429 vnffgr
= self
._vnffgmgr
.create_vnffgr(vnffgr_request
,self
._vnffgd
_msg
.classifier
,vnffg_sff_list
)
430 except Exception as e
:
431 self
._log
.exception("VNFFG instantiation failed: %s", str(e
))
432 self
._vnffgr
_state
= VnffgRecordState
.FAILED
433 raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFFGR %s failure" % (self
.id, vnffgr_request
.id))
435 self
._vnffgr
_state
= VnffgRecordState
.INSTANTIATION_PENDING
437 self
._log
.info("Instantiated VNFFGR :%s", vnffgr
)
438 self
._vnffgr
_state
= VnffgRecordState
.ACTIVE
440 self
._log
.info("Invoking update_state to update NSR state for NSR ID: %s", self
._nsr
.id)
441 yield from self
._nsr
.update_state()
443 def vnffgr_in_vnffgrm(self
):
444 """ Is there a VNFR record in VNFM """
445 if (self
._vnffgr
_state
== VnffgRecordState
.ACTIVE
or
446 self
._vnffgr
_state
== VnffgRecordState
.INSTANTIATION_PENDING
or
447 self
._vnffgr
_state
== VnffgRecordState
.FAILED
):
454 """ Terminate this VNFFGR """
455 if not self
.vnffgr_in_vnffgrm():
456 self
._log
.error("Ignoring terminate request for id %s in state %s",
457 self
.id, self
._vnffgr
_state
)
460 self
._log
.info("Terminating VNFFGR id:%s", self
.id)
461 self
._vnffgr
_state
= VnffgRecordState
.TERMINATE_PENDING
463 self
._vnffgmgr
.terminate_vnffgr(self
._vnffgr
_id
)
465 self
._vnffgr
_state
= VnffgRecordState
.TERMINATED
466 self
._log
.debug("Terminated VNFFGR id:%s", self
.id)
469 class VirtualLinkRecord(object):
470 """ Virtual Link Records class"""
471 XPATH
= "D,/vlr:vlr-catalog/vlr:vlr"
474 def create_record(dts
, log
, loop
, project
, nsr_name
, vld_msg
,
475 cloud_account_name
, om_datacenter
, ip_profile
,
476 nsr_id
, restart_mode
=False):
477 """Creates a new VLR object based on the given data.
479 If restart mode is enabled, then we look for existing records in the
480 DTS and create a VLR records using the exiting data(ID)
485 vlr_obj
= VirtualLinkRecord(
499 res_iter
= yield from dts
.query_read(
500 project
.add_project("D,/vlr:vlr-catalog/vlr:vlr"),
501 rwdts
.XactFlag
.MERGE
)
504 response
= yield from fut
505 vlr
= response
.result
507 # Check if the record is already present, if so use the ID of
508 # the existing record. Since the name of the record is uniquely
509 # formed we can use it as a search key!
510 if vlr
.name
== vlr_obj
.name
:
511 vlr_obj
.reset_id(vlr
.id)
516 def __init__(self
, dts
, log
, loop
, project
, nsr_name
, vld_msg
,
517 cloud_account_name
, om_datacenter
, ip_profile
, nsr_id
):
521 self
._project
= project
522 self
._nsr
_name
= nsr_name
523 self
._vld
_msg
= vld_msg
524 self
._cloud
_account
_name
= cloud_account_name
525 self
._om
_datacenter
_name
= om_datacenter
526 self
._assigned
_subnet
= None
527 self
._nsr
_id
= nsr_id
528 self
._ip
_profile
= ip_profile
529 self
._vlr
_id
= str(uuid
.uuid4())
530 self
._state
= VlRecordState
.INIT
531 self
._prev
_state
= None
532 self
._create
_time
= int(time
.time())
536 """ path for this object """
537 return self
._project
.add_project("D,/vlr:vlr-catalog/vlr:vlr[vlr:id = '{}']".
538 format(self
._vlr
_id
))
547 """ Get NSR name for this VL """
552 """ Virtual Link Desciptor """
556 def assigned_subnet(self
):
557 """ Subnet assigned to this VL"""
558 return self
._assigned
_subnet
563 Get the name for this VLR.
564 VLR name is "nsr name:VLD name"
566 if self
.vld_msg
.vim_network_name
:
567 return self
.vld_msg
.vim_network_name
568 elif self
.vld_msg
.name
== "multisite":
569 # This is a temporary hack to identify manually provisioned inter-site network
570 return self
.vld_msg
.name
572 return self
._nsr
_name
+ "." + self
.vld_msg
.name
575 def cloud_account_name(self
):
576 """ Cloud account that this VLR should be created in """
577 return self
._cloud
_account
_name
580 def om_datacenter_name(self
):
581 """ Datacenter that this VLR should be created in """
582 return self
._om
_datacenter
_name
586 """ Get the VLR path from VLR """
587 return (VirtualLinkRecord
.XPATH
+ "[vlr:id = '{}']").format(vlr
.id)
595 def state(self
, value
):
596 """ VLR set state """
600 def prev_state(self
):
601 """ VLR previous state """
602 return self
._prev
_state
605 def prev_state(self
, value
):
606 """ VLR set previous state """
607 self
._prev
_state
= value
611 """ Virtual Link Record message for Creating VLR in VNS """
612 vld_fields
= ["short_name",
620 vld_copy_dict
= {k
: v
for k
, v
in self
.vld_msg
.as_dict().items()
623 vlr_dict
= {"id": self
._vlr
_id
,
624 "nsr_id_ref": self
._nsr
_id
,
625 "vld_ref": self
.vld_msg
.id,
627 "create_time": self
._create
_time
,
628 "cloud_account": self
.cloud_account_name
,
629 "om_datacenter": self
.om_datacenter_name
,
632 if self
._ip
_profile
and self
._ip
_profile
.has_field('ip_profile_params'):
633 vlr_dict
['ip_profile_params' ] = self
._ip
_profile
.ip_profile_params
.as_dict()
635 vlr_dict
.update(vld_copy_dict
)
636 vlr
= RwVlrYang
.YangData_RwProject_Project_VlrCatalog_Vlr
.from_dict(vlr_dict
)
639 def reset_id(self
, vlr_id
):
640 self
._vlr
_id
= vlr_id
642 def create_nsr_vlr_msg(self
, vnfrs
):
643 """ The VLR message"""
644 nsr_vlr
= RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr_Vlr()
645 nsr_vlr
.vlr_ref
= self
._vlr
_id
646 nsr_vlr
.assigned_subnet
= self
.assigned_subnet
647 nsr_vlr
.cloud_account
= self
.cloud_account_name
648 nsr_vlr
.om_datacenter
= self
.om_datacenter_name
650 for conn
in self
.vld_msg
.vnfd_connection_point_ref
:
652 if (vnfr
.vnfd
.id == conn
.vnfd_id_ref
and
653 vnfr
.member_vnf_index
== conn
.member_vnf_index_ref
and
654 self
.cloud_account_name
== vnfr
.cloud_account_name
and
655 self
.om_datacenter_name
== vnfr
.om_datacenter_name
):
656 cp_entry
= nsr_vlr
.vnfr_connection_point_ref
.add()
657 cp_entry
.vnfr_id
= vnfr
.id
658 cp_entry
.connection_point
= conn
.vnfd_connection_point_ref
663 def instantiate(self
):
664 """ Instantiate this VL """
665 self
._log
.debug("Instaniating VLR key %s, vld %s",
666 self
.xpath
, self
._vld
_msg
)
668 self
._state
= VlRecordState
.INSTANTIATION_PENDING
669 self
._log
.debug("Executing VL create path:%s msg:%s",
670 self
.xpath
, self
.vlr_msg
)
672 with self
._dts
.transaction(flags
=0) as xact
:
673 block
= xact
.block_create()
674 block
.add_query_create(self
.xpath
, self
.vlr_msg
)
675 self
._log
.debug("Executing VL create path:%s msg:%s",
676 self
.xpath
, self
.vlr_msg
)
677 res_iter
= yield from block
.execute(now
=True)
683 self
._state
= VlRecordState
.FAILED
684 raise NsrInstantiationFailed("Failed NS %s instantiation due to empty response" % self
.id)
686 if vlr
.operational_status
== 'failed':
687 self
._log
.debug("NS Id:%s VL creation failed for vlr id %s", self
.id, vlr
.id)
688 self
._state
= VlRecordState
.FAILED
689 raise NsrInstantiationFailed("Failed VL %s instantiation (%s)" % (vlr
.id, vlr
.operational_status_details
))
691 self
._log
.info("Instantiated VL with xpath %s and vlr:%s",
693 self
._state
= VlRecordState
.ACTIVE
694 self
._assigned
_subnet
= vlr
.assigned_subnet
696 def vlr_in_vns(self
):
697 """ Is there a VLR record in VNS """
698 if (self
._state
== VlRecordState
.ACTIVE
or
699 self
._state
== VlRecordState
.INSTANTIATION_PENDING
or
700 self
._state
== VlRecordState
.TERMINATE_PENDING
or
701 self
._state
== VlRecordState
.FAILED
):
708 """ Terminate this VL """
709 if not self
.vlr_in_vns():
710 self
._log
.debug("Ignoring terminate request for id %s in state %s",
711 self
.id, self
._state
)
714 self
._log
.debug("Terminating VL id:%s", self
.id)
715 self
._state
= VlRecordState
.TERMINATE_PENDING
717 with self
._dts
.transaction(flags
=0) as xact
:
718 block
= xact
.block_create()
719 block
.add_query_delete(self
.xpath
)
720 yield from block
.execute(flags
=0, now
=True)
722 self
._state
= VlRecordState
.TERMINATED
723 self
._log
.debug("Terminated VL id:%s", self
.id)
726 class VnfRecordState(Enum
):
727 """ Vnf Record State """
729 INSTANTIATION_PENDING
= 102
731 TERMINATE_PENDING
= 104
736 class VirtualNetworkFunctionRecord(object):
737 """ Virtual Network Function Record class"""
738 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
742 def create_record(dts
, log
, loop
, project
, vnfd
, const_vnfd_msg
, nsd_id
, nsr_name
,
743 cloud_account_name
, om_datacenter_name
, nsr_id
, group_name
, group_instance_id
,
744 placement_groups
, restart_mode
=False):
745 """Creates a new VNFR object based on the given data.
747 If restart mode is enabled, then we look for existing records in the
748 DTS and create a VNFR records using the exiting data(ID)
751 VirtualNetworkFunctionRecord
753 vnfr_obj
= VirtualNetworkFunctionRecord(
768 restart_mode
=restart_mode
)
771 res_iter
= yield from dts
.query_read(
772 project
.add_project("D,/vnfr:vnfr-catalog/vnfr:vnfr"),
773 rwdts
.XactFlag
.MERGE
)
776 response
= yield from fut
777 vnfr
= response
.result
779 if vnfr
.name
== vnfr_obj
.name
:
780 vnfr_obj
.reset_id(vnfr
.id)
798 group_instance_id
=None,
799 placement_groups
= [],
800 restart_mode
= False):
804 self
._project
= project
806 self
._const
_vnfd
_msg
= const_vnfd_msg
807 self
._nsd
_id
= nsd_id
808 self
._nsr
_name
= nsr_name
809 self
._nsr
_id
= nsr_id
810 self
._cloud
_account
_name
= cloud_account_name
811 self
._om
_datacenter
_name
= om_datacenter_name
812 self
._group
_name
= group_name
813 self
._group
_instance
_id
= group_instance_id
814 self
._placement
_groups
= placement_groups
815 self
._config
_status
= NsrYang
.ConfigStates
.INIT
816 self
._create
_time
= int(time
.time())
818 self
._prev
_state
= VnfRecordState
.INIT
819 self
._state
= VnfRecordState
.INIT
820 self
._state
_failed
_reason
= None
822 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
825 self
._vnfr
_id
= str(uuid
.uuid4())
827 self
._vnfr
_msg
= self
.create_vnfr_msg()
828 self
._log
.debug("Set VNFR {} config type to {}".
829 format(self
.name
, self
.config_type
))
830 self
.restart_mode
= restart_mode
833 if group_name
is None and group_instance_id
is not None:
834 raise ValueError("Group instance id must not be provided with an empty group name")
844 return self
._project
.add_project("D,/vnfr:vnfr-catalog/vnfr:vnfr[vnfr:id = '{}']"
850 return self
._vnfr
_msg
853 def const_vnfr_msg(self
):
855 return RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr_ConstituentVnfrRef(
856 vnfr_id
=self
.id, cloud_account
=self
.cloud_account_name
,
857 om_datacenter
=self
._om
_datacenter
_name
)
865 def cloud_account_name(self
):
866 """ Cloud account that this VNF should be created in """
867 return self
._cloud
_account
_name
870 def om_datacenter_name(self
):
871 """ Datacenter that this VNF should be created in """
872 return self
._om
_datacenter
_name
877 """ Is this VNF actve """
878 return True if self
._state
== VnfRecordState
.ACTIVE
else False
882 """ state of this VNF """
886 def state_failed_reason(self
):
887 """ Error message in case this VNF is in failed state """
888 return self
._state
_failed
_reason
891 def member_vnf_index(self
):
892 """ Member VNF index """
893 return self
._const
_vnfd
_msg
.member_vnf_index
898 return self
._nsr
_name
902 """ Name of this VNFR """
903 if self
._name
is not None:
906 name_tags
= [self
._nsr
_name
]
908 if self
._group
_name
is not None:
909 name_tags
.append(self
._group
_name
)
911 if self
._group
_instance
_id
is not None:
912 name_tags
.append(str(self
._group
_instance
_id
))
914 name_tags
.extend([self
.vnfd
.name
, str(self
.member_vnf_index
)])
916 self
._name
= "__".join(name_tags
)
921 def vnfr_xpath(vnfr
):
922 """ Get the VNFR path from VNFR """
923 return (self
._project
.add_project(VirtualNetworkFunctionRecord
.XPATH
) +
924 "[vnfr:id = '{}']").format(vnfr
.id)
927 def config_type(self
):
928 cfg_types
= ['netconf', 'juju', 'script']
929 for method
in cfg_types
:
930 if self
._vnfd
.vnf_configuration
.has_field(method
):
935 def config_status(self
):
936 """Return the config status as YANG ENUM string"""
937 self
._log
.debug("Map VNFR {} config status {} ({})".
938 format(self
.name
, self
._config
_status
, self
.config_type
))
939 if self
.config_type
== 'none':
940 return 'config_not_needed'
941 elif self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
943 elif self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
948 def set_state(self
, state
):
949 """ set the state of this object """
950 self
._prev
_state
= self
._state
953 def reset_id(self
, vnfr_id
):
954 self
._vnfr
_id
= vnfr_id
955 self
._vnfr
_msg
= self
.create_vnfr_msg()
958 self
.config_store
.merge_vnfd_config(
961 self
.member_vnf_index
,
964 def create_vnfr_msg(self
):
965 """ VNFR message for this VNFR """
973 vnfd_copy_dict
= {k
: v
for k
, v
in self
._vnfd
.as_dict().items() if k
in vnfd_fields
}
976 "nsr_id_ref": self
._nsr
_id
,
978 "cloud_account": self
._cloud
_account
_name
,
979 "om_datacenter": self
._om
_datacenter
_name
,
980 "config_status": self
.config_status
982 vnfr_dict
.update(vnfd_copy_dict
)
984 vnfr
= RwVnfrYang
.YangData_RwProject_Project_VnfrCatalog_Vnfr
.from_dict(vnfr_dict
)
985 vnfr
.vnfd
= VnfrYang
.YangData_RwProject_Project_VnfrCatalog_Vnfr_Vnfd
. \
986 from_dict(self
.vnfd
.as_dict(),
987 ignore_missing_keys
=True)
988 vnfr
.member_vnf_index_ref
= self
.member_vnf_index
989 vnfr
.vnf_configuration
.from_dict(self
._vnfd
.vnf_configuration
.as_dict())
991 if self
._vnfd
.mgmt_interface
.has_field("port"):
992 vnfr
.mgmt_interface
.port
= self
._vnfd
.mgmt_interface
.port
994 for group_info
in self
._placement
_groups
:
995 group
= vnfr
.placement_groups_info
.add()
996 group
.from_dict(group_info
.as_dict())
998 # UI expects the monitoring param field to exist
999 vnfr
.monitoring_param
= []
1001 self
._log
.debug("Get vnfr_msg for VNFR {} : {}".format(self
.name
, vnfr
))
1005 def update_vnfm(self
):
1006 self
._log
.debug("Send an update to VNFM for VNFR {} with {}".
1007 format(self
.name
, self
.vnfr_msg
))
1008 yield from self
._dts
.query_update(
1010 0, #rwdts.XactFlag.TRACE,
1014 def get_config_status(self
):
1015 """Return the config status as YANG ENUM"""
1016 return self
._config
_status
1019 def set_config_status(self
, status
):
1021 def status_to_string(status
):
1023 NsrYang
.ConfigStates
.INIT
: 'init',
1024 NsrYang
.ConfigStates
.CONFIGURING
: 'configuring',
1025 NsrYang
.ConfigStates
.CONFIG_NOT_NEEDED
: 'config_not_needed',
1026 NsrYang
.ConfigStates
.CONFIGURED
: 'configured',
1027 NsrYang
.ConfigStates
.FAILED
: 'failed',
1030 return status_dc
[status
]
1032 self
._log
.debug("Update VNFR {} from {} ({}) to {}".
1033 format(self
.name
, self
._config
_status
,
1034 self
.config_type
, status
))
1035 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1036 self
._log
.error("Updating already configured VNFR {}".
1040 if self
._config
_status
!= status
:
1042 self
._config
_status
= status
1043 # I don't think this is used. Original implementor can check.
1044 # Caused Exception, so corrected it by status_to_string
1045 # But not sure whats the use of this variable?
1046 self
.vnfr_msg
.config_status
= status_to_string(status
)
1047 except Exception as e
:
1048 self
._log
.error("Exception=%s", str(e
))
1051 self
._log
.debug("Updated VNFR {} status to {}".format(self
.name
, status
))
1053 if self
._config
_status
!= NsrYang
.ConfigStates
.INIT
:
1055 # Publish only after VNFM has the VNFR created
1056 yield from self
.update_vnfm()
1057 except Exception as e
:
1058 self
._log
.error("Exception updating VNFM with new status {} of VNFR {}: {}".
1059 format(status
, self
.name
, e
))
1060 self
._log
.exception(e
)
1062 def is_configured(self
):
1063 if self
.config_type
== 'none':
1066 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURED
:
1072 def instantiate(self
, nsr
):
1073 """ Instantiate this VNFR"""
1075 self
._log
.debug("Instaniating VNFR key %s, vnfd %s",
1076 self
.xpath
, self
._vnfd
)
1078 self
._log
.debug("Create VNF with xpath %s and vnfr %s",
1079 self
.xpath
, self
.vnfr_msg
)
1081 self
.set_state(VnfRecordState
.INSTANTIATION_PENDING
)
1083 def find_vlr_for_cp(conn
):
1084 """ Find VLR for the given connection point """
1085 for vlr
in nsr
.vlrs
:
1086 for vnfd_cp
in vlr
.vld_msg
.vnfd_connection_point_ref
:
1087 if (vnfd_cp
.vnfd_id_ref
== self
._vnfd
.id and
1088 vnfd_cp
.vnfd_connection_point_ref
== conn
.name
and
1089 vnfd_cp
.member_vnf_index_ref
== self
.member_vnf_index
and
1090 vlr
.cloud_account_name
== self
.cloud_account_name
):
1091 self
._log
.debug("Found VLR for cp_name:%s and vnf-index:%d",
1092 conn
.name
, self
.member_vnf_index
)
1096 # For every connection point in the VNFD fill in the identifier
1097 self
._log
.debug("Add connection point for VNF %s: %s",
1098 self
.vnfr_msg
.name
, self
._vnfd
.connection_point
)
1099 for conn_p
in self
._vnfd
.connection_point
:
1100 cpr
= VnfrYang
.YangData_RwProject_Project_VnfrCatalog_Vnfr_ConnectionPoint()
1101 cpr
.name
= conn_p
.name
1102 cpr
.type_yang
= conn_p
.type_yang
1103 if conn_p
.has_field('port_security_enabled'):
1104 cpr
.port_security_enabled
= conn_p
.port_security_enabled
1106 vlr_ref
= find_vlr_for_cp(conn_p
)
1108 msg
= "Failed to find VLR for cp = %s" % conn_p
.name
1109 self
._log
.debug("%s", msg
)
1110 # raise VirtualNetworkFunctionRecordError(msg)
1113 cpr
.vlr_ref
= vlr_ref
.id
1114 self
.vnfr_msg
.connection_point
.append(cpr
)
1115 self
._log
.debug("Connection point [%s] added, vnf id=%s vnfd id=%s",
1116 cpr
, self
.vnfr_msg
.id, self
.vnfr_msg
.vnfd
.id)
1118 if not self
.restart_mode
:
1119 yield from self
._dts
.query_create(self
.xpath
,
1123 yield from self
._dts
.query_update(self
.xpath
,
1127 self
._log
.info("Created VNF with xpath %s and vnfr %s",
1128 self
.xpath
, self
.vnfr_msg
)
1131 def update_state(self
, vnfr_msg
):
1132 """ Update this VNFR"""
1133 if vnfr_msg
.operational_status
== "running":
1134 if self
.vnfr_msg
.operational_status
!= "running":
1135 yield from self
.is_active()
1136 elif vnfr_msg
.operational_status
== "failed":
1137 yield from self
.instantiation_failed(failed_reason
=vnfr_msg
.operational_status_details
)
1140 def is_active(self
):
1141 """ This VNFR is active """
1142 self
._log
.debug("VNFR %s is active", self
._vnfr
_id
)
1143 self
.set_state(VnfRecordState
.ACTIVE
)
1146 def instantiation_failed(self
, failed_reason
=None):
1147 """ This VNFR instantiation failed"""
1148 self
._log
.error("VNFR %s instantiation failed", self
._vnfr
_id
)
1149 self
.set_state(VnfRecordState
.FAILED
)
1150 self
._state
_failed
_reason
= failed_reason
1152 def vnfr_in_vnfm(self
):
1153 """ Is there a VNFR record in VNFM """
1154 if (self
._state
== VnfRecordState
.ACTIVE
or
1155 self
._state
== VnfRecordState
.INSTANTIATION_PENDING
or
1156 self
._state
== VnfRecordState
.FAILED
):
1162 def terminate(self
):
1163 """ Terminate this VNF """
1164 if not self
.vnfr_in_vnfm():
1165 self
._log
.debug("Ignoring terminate request for id %s in state %s",
1166 self
.id, self
._state
)
1169 self
._log
.debug("Terminating VNF id:%s", self
.id)
1170 self
.set_state(VnfRecordState
.TERMINATE_PENDING
)
1171 with self
._dts
.transaction(flags
=0) as xact
:
1172 block
= xact
.block_create()
1173 block
.add_query_delete(self
.xpath
)
1174 yield from block
.execute(flags
=0)
1175 self
.set_state(VnfRecordState
.TERMINATED
)
1176 self
._log
.debug("Terminated VNF id:%s", self
.id)
1179 class NetworkServiceStatus(object):
1180 """ A class representing the Network service's status """
1181 MAX_EVENTS_RECORDED
= 10
1182 """ Network service Status class"""
1183 def __init__(self
, dts
, log
, loop
):
1188 self
._state
= NetworkServiceRecordState
.INIT
1189 self
._events
= deque([])
1192 def create_notification(self
, evt
, evt_desc
, evt_details
):
1193 xp
= "N,/rw-nsr:nsm-notification"
1194 notif
= RwNsrYang
.YangNotif_RwNsr_NsmNotification()
1196 notif
.description
= evt_desc
1197 notif
.details
= evt_details
if evt_details
is not None else None
1199 yield from self
._dts
.query_create(xp
, rwdts
.XactFlag
.ADVISE
, notif
)
1200 self
._log
.info("Notification called by creating dts query: %s", notif
)
1202 def record_event(self
, evt
, evt_desc
, evt_details
):
1203 """ Record an event """
1204 self
._log
.debug("Recording event - evt %s, evt_descr %s len = %s",
1205 evt
, evt_desc
, len(self
._events
))
1206 if len(self
._events
) >= NetworkServiceStatus
.MAX_EVENTS_RECORDED
:
1207 self
._events
.popleft()
1208 self
._events
.append((int(time
.time()), evt
, evt_desc
,
1209 evt_details
if evt_details
is not None else None))
1211 self
._loop
.create_task(self
.create_notification(evt
,evt_desc
,evt_details
))
1213 def set_state(self
, state
):
1214 """ set the state of this status object """
1218 """ Return the state as a yang enum string """
1219 state_to_str_map
= {"INIT": "init",
1220 "VL_INIT_PHASE": "vl_init_phase",
1221 "VNF_INIT_PHASE": "vnf_init_phase",
1222 "VNFFG_INIT_PHASE": "vnffg_init_phase",
1223 "SCALING_GROUP_INIT_PHASE": "scaling_group_init_phase",
1224 "RUNNING": "running",
1225 "SCALING_OUT": "scaling_out",
1226 "SCALING_IN": "scaling_in",
1227 "TERMINATE_RCVD": "terminate_rcvd",
1228 "TERMINATE": "terminate",
1229 "VL_TERMINATE_PHASE": "vl_terminate_phase",
1230 "VNF_TERMINATE_PHASE": "vnf_terminate_phase",
1231 "VNFFG_TERMINATE_PHASE": "vnffg_terminate_phase",
1232 "TERMINATED": "terminated",
1234 "VL_INSTANTIATE": "vl_instantiate",
1235 "VL_TERMINATE": "vl_terminate",
1237 return state_to_str_map
[self
._state
.name
]
1241 """ State of this status object """
1246 """ Network Service Record as a message"""
1249 for entry
in self
._events
:
1250 event
= RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr_OperationalEvents()
1253 event
.timestamp
, event
.event
, event
.description
, event
.details
= entry
1254 event_list
.append(event
)
1258 class NetworkServiceRecord(object):
1259 """ Network service record """
1260 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
1262 def __init__(self
, dts
, log
, loop
, nsm
, nsm_plugin
, nsr_cfg_msg
,
1263 sdn_account_name
, key_pairs
, project
, restart_mode
=False,
1269 self
._nsr
_cfg
_msg
= nsr_cfg_msg
1270 self
._nsm
_plugin
= nsm_plugin
1271 self
._sdn
_account
_name
= sdn_account_name
1272 self
._vlr
_handler
= vlr_handler
1273 self
._project
= project
1276 self
._nsr
_msg
= None
1277 self
._nsr
_regh
= None
1278 self
._key
_pairs
= key_pairs
1283 self
._param
_pools
= {}
1284 self
._scaling
_groups
= {}
1285 self
._create
_time
= int(time
.time())
1286 self
._op
_status
= NetworkServiceStatus(dts
, log
, loop
)
1287 self
._config
_status
= NsrYang
.ConfigStates
.CONFIGURING
1288 self
._config
_status
_details
= None
1290 self
.restart_mode
= restart_mode
1291 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
1292 self
._debug
_running
= False
1293 self
._is
_active
= False
1294 self
._vl
_phase
_completed
= False
1295 self
._vnf
_phase
_completed
= False
1296 self
.vlr_uptime_tasks
= {}
1299 # Initalise the state to init
1300 # The NSR moves through the following transitions
1301 # 1. INIT -> VLS_READY once all the VLs in the NSD are created
1302 # 2. VLS_READY - VNFS_READY when all the VNFs in the NSD are created
1303 # 3. VNFS_READY - READY when the NSR is published
1305 self
.set_state(NetworkServiceRecordState
.INIT
)
1307 self
.substitute_input_parameters
= InputParameterSubstitution(self
._log
, self
._project
)
1310 def nsm_plugin(self
):
1312 return self
._nsm
_plugin
1314 def set_state(self
, state
):
1315 """ Set state for this NSR"""
1316 self
._log
.debug("Setting state to %s", state
)
1317 # We are in init phase and is moving to the next state
1318 # The new state could be a FAILED state or VNF_INIIT_PHASE
1319 if self
.state
== NetworkServiceRecordState
.VL_INIT_PHASE
:
1320 self
._vl
_phase
_completed
= True
1322 if self
.state
== NetworkServiceRecordState
.VNF_INIT_PHASE
:
1323 self
._vnf
_phase
_completed
= True
1325 self
._op
_status
.set_state(state
)
1326 self
._nsm
_plugin
.set_state(self
.id, state
)
1330 """ Get id for this NSR"""
1331 return self
._nsr
_cfg
_msg
.id
1335 """ Name of this network service record """
1336 return self
._nsr
_cfg
_msg
.name
1339 def cloud_account_name(self
):
1340 return self
._nsr
_cfg
_msg
.cloud_account
1343 def om_datacenter_name(self
):
1344 if self
._nsr
_cfg
_msg
.has_field('om_datacenter'):
1345 return self
._nsr
_cfg
_msg
.om_datacenter
1350 """State of this NetworkServiceRecord"""
1351 return self
._op
_status
.state
1355 """ Is this NSR active ?"""
1356 return True if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
else False
1360 """ VLRs associated with this NSR"""
1365 """ VNFRs associated with this NSR"""
1370 """ VNFFGRs associated with this NSR"""
1371 return self
._vnffgrs
1374 def scaling_groups(self
):
1375 """ Scaling groups associated with this NSR """
1376 return self
._scaling
_groups
1379 def param_pools(self
):
1380 """ Parameter value pools associated with this NSR"""
1381 return self
._param
_pools
1384 def nsr_cfg_msg(self
):
1385 return self
._nsr
_cfg
_msg
1388 def nsr_cfg_msg(self
, msg
):
1389 self
._nsr
_cfg
_msg
= msg
1393 """ NSD Protobuf for this NSR """
1394 if self
._nsd
is not None:
1396 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
1401 """ NSD ID for this NSR """
1402 return self
.nsd_msg
.id
1406 ''' Get a new job id for config primitive'''
1411 def config_status(self
):
1412 """ Config status for NSR """
1413 return self
._config
_status
1415 def resolve_placement_group_cloud_construct(self
, input_group
):
1417 Returns the cloud specific construct for placement group
1419 copy_dict
= ['name', 'requirement', 'strategy']
1421 for group_info
in self
._nsr
_cfg
_msg
.nsd_placement_group_maps
:
1422 if group_info
.placement_group_ref
== input_group
.name
:
1423 group
= VnfrYang
.YangData_RwProject_Project_VnfrCatalog_Vnfr_PlacementGroupsInfo()
1424 group_dict
= {k
:v
for k
,v
in
1425 group_info
.as_dict().items() if k
!= 'placement_group_ref'}
1426 for param
in copy_dict
:
1427 group_dict
.update({param
: getattr(input_group
, param
)})
1428 group
.from_dict(group_dict
)
1434 return "NSR(name={}, nsd_id={}, cloud_account={})".format(
1435 self
.name
, self
.nsd_id
, self
.cloud_account_name
1438 def _get_vnfd(self
, vnfd_id
, config_xact
):
1439 """ Fetch vnfd msg for the passed vnfd id """
1440 return self
._nsm
.get_vnfd(vnfd_id
, config_xact
)
1442 def _get_vnfd_cloud_account(self
, vnfd_member_index
):
1443 """ Fetch Cloud Account for the passed vnfd id """
1444 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1445 vim_accounts
= [(vnf
.cloud_account
,vnf
.om_datacenter
) for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map \
1446 if str(vnfd_member_index
) == vnf
.member_vnf_index_ref
]
1447 if vim_accounts
and vim_accounts
[0]:
1448 return vim_accounts
[0]
1449 return (self
.cloud_account_name
,self
.om_datacenter_name
)
1451 def _get_constituent_vnfd_msg(self
, vnf_index
):
1452 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1453 if const_vnfd
.member_vnf_index
== vnf_index
:
1456 raise ValueError("Constituent VNF index %s not found" % vnf_index
)
1458 def record_event(self
, evt
, evt_desc
, evt_details
=None, state
=None):
1459 """ Record an event """
1460 self
._op
_status
.record_event(evt
, evt_desc
, evt_details
)
1461 if state
is not None:
1462 self
.set_state(state
)
1464 def scaling_trigger_str(self
, trigger
):
1465 SCALING_TRIGGER_STRS
= {
1466 NsdYang
.ScalingTrigger
.PRE_SCALE_IN
: 'pre-scale-in',
1467 NsdYang
.ScalingTrigger
.POST_SCALE_IN
: 'post-scale-in',
1468 NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
: 'pre-scale-out',
1469 NsdYang
.ScalingTrigger
.POST_SCALE_OUT
: 'post-scale-out',
1472 return SCALING_TRIGGER_STRS
[trigger
]
1473 except Exception as e
:
1474 self
._log
.error("Scaling trigger mapping error for {} : {}".
1476 self
._log
.exception(e
)
1477 return "Unknown trigger"
1480 def instantiate_vls(self
):
1482 This function instantiates VLs for every VL in this Network Service
1484 self
._log
.debug("Instantiating %d VLs in NSD id %s", len(self
._vlrs
),
1486 for vlr
in self
._vlrs
:
1487 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1488 vlr
.state
= VlRecordState
.ACTIVE
1489 self
.vlr_uptime_tasks
[vlr
.id] = self
._loop
.create_task(self
.vlr_uptime_update(vlr
))
1492 def vlr_uptime_update(self
, vlr
):
1495 vlr_
= RwVlrYang
.YangData_RwProject_Project_VlrCatalog_Vlr
.from_dict({'id': vlr
.id})
1497 vlr_
.uptime
= int(time
.time()) - vlr
._create
_time
1498 xpath
= self
._project
.add_project(VirtualLinkRecord
.vlr_xpath(vlr
))
1499 yield from self
._vlr
_handler
.update(None, xpath
, vlr_
)
1500 yield from asyncio
.sleep(2, loop
=self
._loop
)
1501 except asyncio
.CancelledError
:
1502 self
._log
.debug("Received cancellation request for vlr_uptime_update task")
1503 xpath
= self
._project
.add_project(VirtualLinkRecord
.vlr_xpath(vlr
))
1504 yield from self
._vlr
_handler
.delete(None, xpath
)
1508 def create(self
, config_xact
):
1509 """ Create this network service"""
1510 # Create virtual links for all the external vnf
1511 # connection points in this NS
1512 yield from self
.create_vls()
1514 # Create VNFs in this network service
1515 yield from self
.create_vnfs(config_xact
)
1517 # Create VNFFG for network service
1518 self
.create_vnffgs()
1520 # Create Scaling Groups for each scaling group in NSD
1521 self
.create_scaling_groups()
1523 # Create Parameter Pools
1524 self
.create_param_pools()
1527 def apply_scale_group_config_script(self
, script
, group
, scale_instance
, trigger
, vnfrs
=None):
1528 """ Apply config based on script for scale group """
1531 def add_vnfrs_data(vnfrs_list
):
1532 """ Add as a dict each of the VNFRs data """
1534 for vnfr
in vnfrs_list
:
1535 self
._log
.debug("Add VNFR {} data".format(vnfr
))
1537 vnfr_data
['name'] = vnfr
.name
1538 if trigger
in [NsdYang
.ScalingTrigger
.PRE_SCALE_IN
, NsdYang
.ScalingTrigger
.POST_SCALE_OUT
]:
1539 # Get VNF management and other IPs, etc
1540 opdata
= yield from self
.fetch_vnfr(vnfr
.xpath
)
1541 self
._log
.debug("VNFR {} op data: {}".format(vnfr
.name
, opdata
))
1543 vnfr_data
['rw_mgmt_ip'] = opdata
.mgmt_interface
.ip_address
1544 vnfr_data
['rw_mgmt_port'] = opdata
.mgmt_interface
.port
1545 except Exception as e
:
1546 self
._log
.error("Unable to get management IP for vnfr {}:{}".
1547 format(vnfr
.name
, e
))
1550 vnfr_data
['connection_points'] = []
1551 for cp
in opdata
.connection_point
:
1553 con_pt
['name'] = cp
.name
1554 con_pt
['ip_address'] = cp
.ip_address
1555 vnfr_data
['connection_points'].append(con_pt
)
1556 except Exception as e
:
1557 self
._log
.error("Exception getting connections points for VNFR {}: {}".
1558 format(vnfr
.name
, e
))
1560 vnfrs_data
.append(vnfr_data
)
1561 self
._log
.debug("VNFRs data: {}".format(vnfrs_data
))
1565 def add_nsr_data(nsr
):
1567 nsr_data
['name'] = nsr
.name
1570 if script
is None or len(script
) == 0:
1571 self
._log
.error("Script not provided for scale group config: {}".format(group
.name
))
1574 if script
[0] == '/':
1577 path
= os
.path
.join(os
.environ
['RIFT_INSTALL'], "usr/bin", script
)
1578 if not os
.path
.exists(path
):
1579 self
._log
.error("Config faled for scale group {}: Script does not exist at {}".
1580 format(group
.name
, path
))
1583 # Build a YAML file with all parameters for the script to execute
1584 # The data consists of 5 sections
1586 # 2. Scale group config
1587 # 3. VNFRs in the scale group
1588 # 4. VNFRs outside scale group
1591 data
['trigger'] = group
.trigger_map(trigger
)
1592 data
['config'] = group
.group_msg
.as_dict()
1595 data
["vnfrs_in_group"] = yield from add_vnfrs_data(vnfrs
)
1597 data
["vnfrs_in_group"] = yield from add_vnfrs_data(scale_instance
.vnfrs
)
1599 data
["vnfrs_others"] = yield from add_vnfrs_data(self
.vnfrs
.values())
1600 data
["nsr"] = add_nsr_data(self
)
1603 with tempfile
.NamedTemporaryFile(delete
=False) as tmp_file
:
1604 tmp_file
.write(yaml
.dump(data
, default_flow_style
=True)
1607 self
._log
.debug("Creating a temp file: {} with input data: {}".
1608 format(tmp_file
.name
, data
))
1610 cmd
= "{} {}".format(path
, tmp_file
.name
)
1611 self
._log
.debug("Running the CMD: {}".format(cmd
))
1612 proc
= yield from asyncio
.create_subprocess_shell(cmd
, loop
=self
._loop
)
1613 rc
= yield from proc
.wait()
1615 self
._log
.error("The script {} for scale group {} config returned: {}".
1616 format(script
, group
.name
, rc
))
1624 def apply_scaling_group_config(self
, trigger
, group
, scale_instance
, vnfrs
=None):
1625 """ Apply the config for the scaling group based on trigger """
1626 if group
is None or scale_instance
is None:
1630 def update_config_status(success
=True, err_msg
=None):
1631 self
._log
.debug("Update %s config status to %r : %s",
1632 scale_instance
, success
, err_msg
)
1633 if (scale_instance
.config_status
== "failed"):
1634 # Do not update the config status if it is already in failed state
1637 if scale_instance
.config_status
== "configured":
1638 # Update only to failed state an already configured scale instance
1640 scale_instance
.config_status
= "failed"
1641 scale_instance
.config_err_msg
= err_msg
1642 yield from self
.update_state()
1644 # We are in configuring state
1645 # Only after post scale out mark instance as configured
1646 if trigger
== NsdYang
.ScalingTrigger
.POST_SCALE_OUT
:
1648 scale_instance
.config_status
= "configured"
1650 scale_instance
.config_status
= "failed"
1651 scale_instance
.config_err_msg
= err_msg
1652 yield from self
.update_state()
1654 config
= group
.trigger_config(trigger
)
1658 self
._log
.debug("Scaling group {} config: {}".format(group
.name
, config
))
1659 if config
.has_field("ns_config_primitive_name_ref"):
1660 config_name
= config
.ns_config_primitive_name_ref
1661 nsd_msg
= self
.nsd_msg
1662 config_primitive
= None
1663 for ns_cfg_prim
in nsd_msg
.service_primitive
:
1664 if ns_cfg_prim
.name
== config_name
:
1665 config_primitive
= ns_cfg_prim
1668 if config_primitive
is None:
1669 raise ValueError("Could not find ns_cfg_prim %s in nsr %s" % (config_name
, self
.name
))
1671 self
._log
.debug("Scaling group {} config primitive: {}".format(group
.name
, config_primitive
))
1672 if config_primitive
.has_field("user_defined_script"):
1673 rc
= yield from self
.apply_scale_group_config_script(config_primitive
.user_defined_script
,
1674 group
, scale_instance
, trigger
, vnfrs
)
1677 err_msg
= "Failed config for trigger {} using config script '{}'". \
1678 format(self
.scaling_trigger_str(trigger
),
1679 config_primitive
.user_defined_script
)
1680 yield from update_config_status(success
=rc
, err_msg
=err_msg
)
1683 err_msg
= "Failed config for trigger {} as config script is not specified". \
1684 format(self
.scaling_trigger_str(trigger
))
1685 yield from update_config_status(success
=False, err_msg
=err_msg
)
1686 raise NotImplementedError("Only script based config support for scale group for now: {}".
1689 err_msg
= "Failed config for trigger {} as config primitive is not specified".\
1690 format(self
.scaling_trigger_str(trigger
))
1691 yield from update_config_status(success
=False, err_msg
=err_msg
)
1692 self
._log
.error("Config primitive not specified for config action in scale group %s" %
1696 def create_scaling_groups(self
):
1697 """ This function creates a NSScalingGroup for every scaling
1698 group defined in he NSD"""
1700 for scaling_group_msg
in self
.nsd_msg
.scaling_group_descriptor
:
1701 self
._log
.debug("Found scaling_group %s in nsr id %s",
1702 scaling_group_msg
.name
, self
.id)
1704 group_record
= scale_group
.ScalingGroup(
1709 self
._scaling
_groups
[group_record
.name
] = group_record
1712 def create_scale_group_instance(self
, group_name
, index
, config_xact
, is_default
=False):
1713 group
= self
._scaling
_groups
[group_name
]
1714 scale_instance
= group
.create_instance(index
, is_default
)
1718 self
._log
.debug("Creating %u VNFs associated with NS id %s scaling group %s",
1719 len(self
.nsd_msg
.constituent_vnfd
), self
.id, self
)
1722 for vnf_index
, count
in group
.vnf_index_count_map
.items():
1723 const_vnfd_msg
= self
._get
_constituent
_vnfd
_msg
(vnf_index
)
1724 vnfd_msg
= self
._get
_vnfd
(const_vnfd_msg
.vnfd_id_ref
, config_xact
)
1726 cloud_account_name
, om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd_msg
.member_vnf_index
)
1727 if cloud_account_name
is None:
1728 cloud_account_name
= self
.cloud_account_name
1729 for _
in range(count
):
1730 vnfr
= yield from self
.create_vnf_record(vnfd_msg
, const_vnfd_msg
, cloud_account_name
, om_datacenter_name
, group_name
, index
)
1731 scale_instance
.add_vnfr(vnfr
)
1737 def instantiate_instance():
1738 self
._log
.debug("Creating %s VNFRS", scale_instance
)
1739 vnfrs
= yield from create_vnfs()
1740 yield from self
.publish()
1742 self
._log
.debug("Instantiating %s VNFRS for %s", len(vnfrs
), scale_instance
)
1743 scale_instance
.operational_status
= "vnf_init_phase"
1744 yield from self
.update_state()
1747 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
,
1748 group
, scale_instance
, vnfrs
)
1750 self
._log
.error("Pre scale out config for scale group {} ({}) failed".
1751 format(group
.name
, index
))
1752 scale_instance
.operational_status
= "failed"
1754 yield from self
.instantiate_vnfs(vnfrs
)
1756 except Exception as e
:
1757 self
._log
.exception("Failed to begin instantiatiation of vnfs for scale group {}: {}".
1758 format(group
.name
, e
))
1759 self
._log
.exception(e
)
1760 scale_instance
.operational_status
= "failed"
1762 yield from self
.update_state()
1764 yield from instantiate_instance()
1767 def delete_scale_group_instance(self
, group_name
, index
):
1768 group
= self
._scaling
_groups
[group_name
]
1769 scale_instance
= group
.get_instance(index
)
1770 if scale_instance
.is_default
:
1771 raise ScalingOperationError("Cannot terminate a default scaling group instance")
1773 scale_instance
.operational_status
= "terminate"
1774 yield from self
.update_state()
1777 def terminate_instance():
1778 self
._log
.debug("Terminating %s VNFRS" % scale_instance
)
1779 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_IN
,
1780 group
, scale_instance
)
1782 self
._log
.error("Pre scale in config for scale group {} ({}) failed".
1783 format(group
.name
, index
))
1785 # Going ahead with terminate, even if there is an error in pre-scale-in config
1786 # as this could be result of scale out failure and we need to cleanup this group
1787 yield from self
.terminate_vnfrs(scale_instance
.vnfrs
)
1788 group
.delete_instance(index
)
1790 scale_instance
.operational_status
= "vnf_terminate_phase"
1791 yield from self
.update_state()
1793 yield from terminate_instance()
1796 def _update_scale_group_instances_status(self
):
1798 def post_scale_out_task(group
, instance
):
1799 # Apply post scale out config once all VNFRs are active
1800 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_OUT
,
1802 instance
.operational_status
= "running"
1804 self
._log
.debug("Scale out for group {} and instance {} succeeded".
1805 format(group
.name
, instance
.instance_id
))
1807 self
._log
.error("Post scale out config for scale group {} ({}) failed".
1808 format(group
.name
, instance
.instance_id
))
1810 yield from self
.update_state()
1812 group_instances
= {group
: group
.instances
for group
in self
._scaling
_groups
.values()}
1813 for group
, instances
in group_instances
.items():
1814 self
._log
.debug("Updating %s instance status", group
)
1815 for instance
in instances
:
1816 instance_vnf_state_list
= [vnfr
.state
for vnfr
in instance
.vnfrs
]
1817 self
._log
.debug("Got vnfr instance states: %s", instance_vnf_state_list
)
1818 if instance
.operational_status
== "vnf_init_phase":
1819 if all([state
== VnfRecordState
.ACTIVE
for state
in instance_vnf_state_list
]):
1820 instance
.operational_status
= "running"
1822 # Create a task for post scale out to allow us to sleep before attempting
1823 # to configure newly created VM's
1824 self
._loop
.create_task(post_scale_out_task(group
, instance
))
1826 elif any([state
== VnfRecordState
.FAILED
for state
in instance_vnf_state_list
]):
1827 self
._log
.debug("Scale out for group {} and instance {} failed".
1828 format(group
.name
, instance
.instance_id
))
1829 instance
.operational_status
= "failed"
1831 elif instance
.operational_status
== "vnf_terminate_phase":
1832 if all([state
== VnfRecordState
.TERMINATED
for state
in instance_vnf_state_list
]):
1833 instance
.operational_status
= "terminated"
1834 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_IN
,
1837 self
._log
.debug("Scale in for group {} and instance {} succeeded".
1838 format(group
.name
, instance
.instance_id
))
1840 self
._log
.error("Post scale in config for scale group {} ({}) failed".
1841 format(group
.name
, instance
.instance_id
))
1843 def create_vnffgs(self
):
1844 """ This function creates VNFFGs for every VNFFG in the NSD
1845 associated with this NSR"""
1847 for vnffgd
in self
.nsd_msg
.vnffgd
:
1848 self
._log
.debug("Found vnffgd %s in nsr id %s", vnffgd
, self
.id)
1849 vnffgr
= VnffgRecord(self
._dts
,
1852 self
._nsm
._vnffgmgr
,
1856 self
._sdn
_account
_name
1858 self
._vnffgrs
[vnffgr
.id] = vnffgr
1860 def resolve_vld_ip_profile(self
, nsd_msg
, vld
):
1861 self
._log
.debug("Receieved ip profile ref is %s",vld
.ip_profile_ref
)
1862 if not vld
.has_field('ip_profile_ref'):
1864 profile
= [profile
for profile
in nsd_msg
.ip_profiles
if profile
.name
== vld
.ip_profile_ref
]
1865 return profile
[0] if profile
else None
1868 def _create_vls(self
, vld
, cloud_account
,om_datacenter
):
1869 """Create a VLR in the cloud account specified using the given VLD
1873 cloud_account : Cloud account name
1878 vlr
= yield from VirtualLinkRecord
.create_record(
1887 self
.resolve_vld_ip_profile(self
.nsd_msg
, vld
),
1889 restart_mode
=self
.restart_mode
)
1893 def _extract_cloud_accounts_for_vl(self
, vld
):
1895 Extracts the list of cloud accounts from the NS Config obj
1898 1. Cloud accounts based connection point (vnf_cloud_account_map)
1900 vld : VLD yang object
1905 cloud_account_list
= []
1907 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1908 # Handle case where cloud_account is None
1910 for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1911 if vnf
.cloud_account
is not None or vnf
.om_datacenter
is not None:
1912 vnf_cloud_map
[vnf
.member_vnf_index_ref
] = (vnf
.cloud_account
,vnf
.om_datacenter
)
1914 for vnfc
in vld
.vnfd_connection_point_ref
:
1915 cloud_account
= vnf_cloud_map
.get(
1916 vnfc
.member_vnf_index_ref
,
1917 (self
.cloud_account_name
,self
.om_datacenter_name
))
1919 cloud_account_list
.append(cloud_account
)
1921 if self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1922 for vld_map
in self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1923 if vld_map
.vld_id_ref
== vld
.id:
1924 for cloud_account
in vld_map
.cloud_accounts
:
1925 cloud_account_list
.extend((cloud_account
,None))
1926 for om_datacenter
in vld_map
.om_datacenters
:
1927 cloud_account_list
.extend((None,om_datacenter
))
1929 # If no config has been provided then fall-back to the default
1931 if not cloud_account_list
:
1932 cloud_account_list
= [(self
.cloud_account_name
,self
.om_datacenter_name
)]
1934 self
._log
.debug("VL {} cloud accounts: {}".
1935 format(vld
.name
, cloud_account_list
))
1936 return set(cloud_account_list
)
1939 def create_vls(self
):
1940 """ This function creates VLs for every VLD in the NSD
1941 associated with this NSR"""
1942 for vld
in self
.nsd_msg
.vld
:
1944 self
._log
.debug("Found vld %s in nsr id %s", vld
, self
.id)
1945 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1946 for cloud_account
,om_datacenter
in cloud_account_list
:
1947 vlr
= yield from self
._create
_vls
(vld
, cloud_account
,om_datacenter
)
1948 self
._vlrs
.append(vlr
)
1952 def create_vl_instance(self
, vld
):
1953 self
._log
.debug("Create VL for {}: {}".format(self
.id, vld
.as_dict()))
1954 # Check if the VL is already present
1956 for vl
in self
._vlrs
:
1957 if vl
.vld_msg
.id == vld
.id:
1958 self
._log
.debug("The VLD %s already in NSR %s as VLR %s with status %s",
1959 vld
.id, self
.id, vl
.id, vl
.state
)
1961 if vlr
.state
!= VlRecordState
.TERMINATED
:
1962 err_msg
= "VLR for VL %s in NSR %s already instantiated", \
1964 self
._log
.error(err_msg
)
1965 raise NsrVlUpdateError(err_msg
)
1969 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1970 for account
,om_datacenter
in cloud_account_list
:
1971 vlr
= yield from self
._create
_vls
(vld
, account
,om_datacenter
)
1972 self
._vlrs
.append(vlr
)
1974 vlr
.state
= VlRecordState
.INSTANTIATION_PENDING
1975 yield from self
.update_state()
1978 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1979 vlr
.state
= VlRecordState
.ACTIVE
1981 except Exception as e
:
1982 err_msg
= "Error instantiating VL for NSR {} and VLD {}: {}". \
1983 format(self
.id, vld
.id, e
)
1984 self
._log
.error(err_msg
)
1985 self
._log
.exception(e
)
1986 vlr
.state
= VlRecordState
.FAILED
1988 yield from self
.update_state()
1991 def delete_vl_instance(self
, vld
):
1992 for vlr
in self
._vlrs
:
1993 if vlr
.vld_msg
.id == vld
.id:
1994 self
._log
.debug("Found VLR %s for VLD %s in NSR %s",
1995 vlr
.id, vld
.id, self
.id)
1996 vlr
.state
= VlRecordState
.TERMINATE_PENDING
1997 yield from self
.update_state()
2000 yield from self
.nsm_plugin
.terminate_vl(vlr
)
2001 vlr
.state
= VlRecordState
.TERMINATED
2002 self
._vlrs
.remove(vlr
)
2004 except Exception as e
:
2005 err_msg
= "Error terminating VL for NSR {} and VLD {}: {}". \
2006 format(self
.id, vld
.id, e
)
2007 self
._log
.error(err_msg
)
2008 self
._log
.exception(e
)
2009 vlr
.state
= VlRecordState
.FAILED
2011 yield from self
.update_state()
2015 def create_vnfs(self
, config_xact
):
2017 This function creates VNFs for every VNF in the NSD
2018 associated with this NSR
2020 self
._log
.debug("Creating %u VNFs associated with this NS id %s",
2021 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
2023 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
2024 if not const_vnfd
.start_by_default
:
2025 self
._log
.debug("start_by_default set to False in constituent VNF (%s). Skipping start.",
2026 const_vnfd
.member_vnf_index
)
2029 vnfd_msg
= self
._get
_vnfd
(const_vnfd
.vnfd_id_ref
, config_xact
)
2030 cloud_account_name
,om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd
.member_vnf_index
)
2031 if cloud_account_name
is None:
2032 cloud_account_name
= self
.cloud_account_name
2033 yield from self
.create_vnf_record(vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
)
2036 def get_placement_groups(self
, vnfd_msg
, const_vnfd
):
2037 placement_groups
= []
2038 for group
in self
.nsd_msg
.placement_groups
:
2039 for member_vnfd
in group
.member_vnfd
:
2040 if (member_vnfd
.vnfd_id_ref
== vnfd_msg
.id) and \
2041 (member_vnfd
.member_vnf_index_ref
== str(const_vnfd
.member_vnf_index
)):
2042 group_info
= self
.resolve_placement_group_cloud_construct(group
)
2043 if group_info
is None:
2044 self
._log
.error("Could not resolve cloud-construct for placement group: %s", group
.name
)
2045 ### raise PlacementGroupError("Could not resolve cloud-construct for placement group: {}".format(group.name))
2047 self
._log
.info("Successfully resolved cloud construct for placement group: %s for VNF: %s (Member Index: %s)",
2050 const_vnfd
.member_vnf_index
)
2051 placement_groups
.append(group_info
)
2052 return placement_groups
2055 def create_vnf_record(self
, vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
, group_name
=None, group_instance_id
=None):
2056 # Fetch the VNFD associated with this VNF
2057 placement_groups
= self
.get_placement_groups(vnfd_msg
, const_vnfd
)
2058 self
._log
.info("Cloud Account for VNF %d is %s",const_vnfd
.member_vnf_index
,cloud_account_name
)
2059 self
._log
.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s, restart mode self.restart_mode %s",
2061 const_vnfd
.member_vnf_index
,
2062 [ group
.name
for group
in placement_groups
],
2064 vnfr
= yield from VirtualNetworkFunctionRecord
.create_record(self
._dts
,
2078 restart_mode
=self
.restart_mode
,
2080 if vnfr
.id in self
._vnfrs
:
2081 err
= "VNF with VNFR id %s already in vnf list" % (vnfr
.id,)
2082 raise NetworkServiceRecordError(err
)
2084 self
._vnfrs
[vnfr
.id] = vnfr
2085 self
._nsm
.vnfrs
[vnfr
.id] = vnfr
2087 yield from vnfr
.set_config_status(NsrYang
.ConfigStates
.INIT
)
2089 self
._log
.debug("Added VNFR %s to NSM VNFR list with id %s",
2095 def create_param_pools(self
):
2096 for param_pool
in self
.nsd_msg
.parameter_pool
:
2097 self
._log
.debug("Found parameter pool %s in nsr id %s", param_pool
, self
.id)
2099 start_value
= param_pool
.range.start_value
2100 end_value
= param_pool
.range.end_value
2101 if end_value
< start_value
:
2102 raise NetworkServiceRecordError(
2103 "Parameter pool %s has invalid range (start: {}, end: {})".format(
2104 start_value
, end_value
2108 self
._param
_pools
[param_pool
.name
] = config_value_pool
.ParameterValuePool(
2111 range(start_value
, end_value
)
2115 def fetch_vnfr(self
, vnfr_path
):
2116 """ Fetch VNFR record """
2118 self
._log
.debug("Fetching VNFR with key %s while instantiating %s",
2120 res_iter
= yield from self
._dts
.query_read(vnfr_path
, rwdts
.XactFlag
.MERGE
)
2122 for ent
in res_iter
:
2123 res
= yield from ent
2129 def instantiate_vnfs(self
, vnfrs
):
2131 This function instantiates VNFs for every VNF in this Network Service
2133 self
._log
.debug("Instantiating %u VNFs in NS %s", len(vnfrs
), self
.id)
2135 self
._log
.debug("Instantiating VNF: %s in NS %s", vnf
, self
.id)
2136 yield from self
.nsm_plugin
.instantiate_vnf(self
, vnf
)
2139 def instantiate_vnffgs(self
):
2141 This function instantiates VNFFGs for every VNFFG in this Network Service
2143 self
._log
.debug("Instantiating %u VNFFGs in NS %s",
2144 len(self
.nsd_msg
.vnffgd
), self
.id)
2145 for _
, vnfr
in self
.vnfrs
.items():
2146 while vnfr
.state
in [VnfRecordState
.INSTANTIATION_PENDING
, VnfRecordState
.INIT
]:
2147 self
._log
.debug("Received vnfr state for vnfr %s is %s; retrying",vnfr
.name
,vnfr
.state
)
2148 yield from asyncio
.sleep(2, loop
=self
._loop
)
2149 if vnfr
.state
== VnfRecordState
.ACTIVE
:
2150 self
._log
.debug("Received vnfr state for vnfr %s is %s ",vnfr
.name
,vnfr
.state
)
2153 self
._log
.debug("Received vnfr state for vnfr %s is %s; failing vnffg creation",vnfr
.name
,vnfr
.state
)
2154 self
._vnffgr
_state
= VnffgRecordState
.FAILED
2157 self
._log
.info("Waiting for 90 seconds for VMs to come up")
2158 yield from asyncio
.sleep(90, loop
=self
._loop
)
2159 self
._log
.info("Starting VNFFG orchestration")
2160 for vnffg
in self
._vnffgrs
.values():
2161 self
._log
.debug("Instantiating VNFFG: %s in NS %s", vnffg
, self
.id)
2162 yield from vnffg
.instantiate()
2165 def instantiate_scaling_instances(self
, config_xact
):
2166 """ Instantiate any default scaling instances in this Network Service """
2167 for group
in self
._scaling
_groups
.values():
2168 for i
in range(group
.min_instance_count
):
2169 self
._log
.debug("Instantiating %s default scaling instance %s", group
, i
)
2170 yield from self
.create_scale_group_instance(
2171 group
.name
, i
, config_xact
, is_default
=True
2174 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2175 if group_msg
.scaling_group_name_ref
!= group
.name
:
2178 for instance
in group_msg
.instance
:
2179 self
._log
.debug("Reloading %s scaling instance %s", group_msg
, instance
.id)
2180 yield from self
.create_scale_group_instance(
2181 group
.name
, instance
.id, config_xact
, is_default
=False
2184 def has_scaling_instances(self
):
2185 """ Return boolean indicating if the network service has default scaling groups """
2186 for group
in self
._scaling
_groups
.values():
2187 if group
.min_instance_count
> 0:
2190 for group_msg
in self
._nsr
_cfg
_msg
.scaling_group
:
2191 if len(group_msg
.instance
) > 0:
2198 """ This function publishes this NSR """
2199 self
._nsr
_msg
= self
.create_msg()
2201 self
._log
.debug("Publishing the NSR with xpath %s and nsr %s",
2205 if self
._debug
_running
:
2206 self
._log
.debug("Publishing NSR in RUNNING state!")
2209 with self
._dts
.transaction() as xact
:
2210 yield from self
._nsm
.nsr_handler
.update(xact
, self
.nsr_xpath
, self
._nsr
_msg
)
2211 if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
:
2212 self
._debug
_running
= True
2215 def unpublish(self
, xact
):
2216 """ Unpublish this NSR object """
2217 self
._log
.debug("Unpublishing Network service id %s", self
.id)
2218 yield from self
._nsm
.nsr_handler
.delete(xact
, self
.nsr_xpath
)
2221 def nsr_xpath(self
):
2222 """ Returns the xpath associated with this NSR """
2223 return self
._project
.add_project((
2224 "D,/nsr:ns-instance-opdata" +
2225 "/nsr:nsr[nsr:ns-instance-config-ref = '{}']"
2229 def xpath_from_nsr(nsr
):
2230 """ Returns the xpath associated with this NSR op data"""
2231 return self
._project
.add_project((NetworkServiceRecord
.XPATH
+
2232 "[nsr:ns-instance-config-ref = '{}']").format(nsr
.id))
2235 def nsd_xpath(self
):
2236 """ Return NSD config xpath."""
2237 return self
._project
.add_project((
2238 "C,/project-nsd:nsd-catalog/project-nsd:nsd[project-nsd:id = '{}']"
2239 ).format(self
.nsd_id
))
2242 def instantiate(self
, config_xact
):
2243 """"Instantiates a NetworkServiceRecord.
2245 This function instantiates a Network service
2246 which involves the following steps,
2248 * Instantiate every VL in NSD by sending create VLR request to DTS.
2249 * Instantiate every VNF in NSD by sending create VNF reuqest to DTS.
2250 * Publish the NSR details to DTS
2253 nsr: The NSR configuration request containing nsr-id and nsd
2254 config_xact: The configuration transaction which initiated the instatiation
2257 NetworkServiceRecordError if the NSR creation fails
2263 self
._log
.debug("Instantiating NS - %s xact - %s", self
, config_xact
)
2265 # Move the state to INIITALIZING
2266 self
.set_state(NetworkServiceRecordState
.INIT
)
2268 event_descr
= "Instantiation Request Received NSR Id:%s" % self
.id
2269 self
.record_event("instantiating", event_descr
)
2272 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
2275 # Update ref count if nsd present in catalog
2276 self
._nsm
.get_nsd_ref(self
.nsd_id
)
2278 except NetworkServiceDescriptorError
:
2279 # This could be an NSD not in the nsd-catalog
2282 # Merge any config and initial config primitive values
2283 self
.config_store
.merge_nsd_config(self
.nsd_msg
)
2284 self
._log
.debug("Merged NSD: {}".format(self
.nsd_msg
.as_dict()))
2286 event_descr
= "Fetched NSD with descriptor id %s" % self
.nsd_id
2287 self
.record_event("nsd-fetched", event_descr
)
2289 if self
._nsd
is None:
2290 msg
= "Failed to fetch NSD with nsd-id [%s] for nsr-id %s"
2291 self
._log
.debug(msg
, self
.nsd_id
, self
.id)
2292 raise NetworkServiceRecordError(self
)
2294 self
._log
.debug("Got nsd result %s", self
._nsd
)
2296 # Substitute any input parameters
2297 self
.substitute_input_parameters(self
._nsd
, self
._nsr
_cfg
_msg
)
2300 yield from self
.create(config_xact
)
2302 # Publish the NSR to DTS
2303 yield from self
.publish()
2306 def do_instantiate():
2308 Instantiate network service
2310 self
._log
.debug("Instantiating VLs nsr id [%s] nsd id [%s]",
2311 self
.id, self
.nsd_id
)
2313 # instantiate the VLs
2314 event_descr
= ("Instantiating %s external VLs for NSR id %s" %
2315 (len(self
.nsd_msg
.vld
), self
.id))
2316 self
.record_event("begin-external-vls-instantiation", event_descr
)
2318 self
.set_state(NetworkServiceRecordState
.VL_INIT_PHASE
)
2320 yield from self
.instantiate_vls()
2322 # Publish the NSR to DTS
2323 yield from self
.publish()
2325 event_descr
= ("Finished instantiating %s external VLs for NSR id %s" %
2326 (len(self
.nsd_msg
.vld
), self
.id))
2327 self
.record_event("end-external-vls-instantiation", event_descr
)
2329 self
.set_state(NetworkServiceRecordState
.VNF_INIT_PHASE
)
2331 self
._log
.debug("Instantiating VNFs ...... nsr[%s], nsd[%s]",
2332 self
.id, self
.nsd_id
)
2334 # instantiate the VNFs
2335 event_descr
= ("Instantiating %s VNFS for NSR id %s" %
2336 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2338 self
.record_event("begin-vnf-instantiation", event_descr
)
2340 yield from self
.instantiate_vnfs(self
._vnfrs
.values())
2342 self
._log
.debug(" Finished instantiating %d VNFs for NSR id %s",
2343 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
2345 event_descr
= ("Finished instantiating %s VNFs for NSR id %s" %
2346 (len(self
.nsd_msg
.constituent_vnfd
), self
.id))
2347 self
.record_event("end-vnf-instantiation", event_descr
)
2349 if len(self
.vnffgrs
) > 0:
2350 #self.set_state(NetworkServiceRecordState.VNFFG_INIT_PHASE)
2351 event_descr
= ("Instantiating %s VNFFGS for NSR id %s" %
2352 (len(self
.nsd_msg
.vnffgd
), self
.id))
2354 self
.record_event("begin-vnffg-instantiation", event_descr
)
2356 yield from self
.instantiate_vnffgs()
2358 event_descr
= ("Finished instantiating %s VNFFGDs for NSR id %s" %
2359 (len(self
.nsd_msg
.vnffgd
), self
.id))
2360 self
.record_event("end-vnffg-instantiation", event_descr
)
2362 if self
.has_scaling_instances():
2363 event_descr
= ("Instantiating %s Scaling Groups for NSR id %s" %
2364 (len(self
._scaling
_groups
), self
.id))
2366 self
.record_event("begin-scaling-group-instantiation", event_descr
)
2367 yield from self
.instantiate_scaling_instances(config_xact
)
2368 self
.record_event("end-scaling-group-instantiation", event_descr
)
2370 # Give the plugin a chance to deploy the network service now that all
2371 # virtual links and vnfs are instantiated
2372 yield from self
.nsm_plugin
.deploy(self
._nsr
_msg
)
2374 self
._log
.debug("Publishing NSR...... nsr[%s], nsd[%s]",
2375 self
.id, self
.nsd_id
)
2377 # Publish the NSR to DTS
2378 yield from self
.publish()
2380 self
._log
.debug("Published NSR...... nsr[%s], nsd[%s]",
2381 self
.id, self
.nsd_id
)
2383 def on_instantiate_done(fut
):
2384 # If the do_instantiate fails, then publish NSR with failed result
2385 if fut
.exception() is not None:
2386 self
._log
.error("NSR instantiation failed for NSR id %s: %s", self
.id, str(fut
.exception()))
2387 self
._loop
.create_task(self
.instantiation_failed(failed_reason
=str(fut
.exception())))
2389 instantiate_task
= self
._loop
.create_task(do_instantiate())
2390 instantiate_task
.add_done_callback(on_instantiate_done
)
2393 def set_config_status(self
, status
, status_details
=None):
2394 if self
.config_status
!= status
:
2395 self
._log
.debug("Updating NSR {} status for {} to {}".
2396 format(self
.name
, self
.config_status
, status
))
2397 self
._config
_status
= status
2398 self
._config
_status
_details
= status_details
2400 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2401 self
.record_event("config-failed", "NS configuration failed",
2402 evt_details
=self
._config
_status
_details
)
2404 yield from self
.publish()
2407 def is_active(self
):
2408 """ This NS is active """
2409 self
.set_state(NetworkServiceRecordState
.RUNNING
)
2413 # Publish the NSR to DTS
2414 self
._log
.debug("Network service %s is active ", self
.id)
2415 self
._is
_active
= True
2417 event_descr
= "NSR in running state for NSR id %s" % self
.id
2418 self
.record_event("ns-running", event_descr
)
2420 yield from self
.publish()
2423 def instantiation_failed(self
, failed_reason
=None):
2424 """ The NS instantiation failed"""
2425 self
._log
.error("Network service id:%s, name:%s instantiation failed",
2427 self
.set_state(NetworkServiceRecordState
.FAILED
)
2429 event_descr
= "Instantiation of NS %s failed" % self
.id
2430 self
.record_event("ns-failed", event_descr
, evt_details
=failed_reason
)
2432 # Publish the NSR to DTS
2433 yield from self
.publish()
2436 def terminate_vnfrs(self
, vnfrs
):
2437 """ Terminate VNFRS in this network service """
2438 self
._log
.debug("Terminating VNFs in network service %s", self
.id)
2440 yield from self
.nsm_plugin
.terminate_vnf(vnfr
)
2443 def terminate(self
):
2444 """ Terminate a NetworkServiceRecord."""
2445 def terminate_vnffgrs():
2446 """ Terminate VNFFGRS in this network service """
2447 self
._log
.debug("Terminating VNFFGRs in network service %s", self
.id)
2448 for vnffgr
in self
.vnffgrs
.values():
2449 yield from vnffgr
.terminate()
2451 def terminate_vlrs():
2452 """ Terminate VLRs in this netork service """
2453 self
._log
.debug("Terminating VLs in network service %s", self
.id)
2454 for vlr
in self
.vlrs
:
2455 yield from self
.nsm_plugin
.terminate_vl(vlr
)
2456 vlr
.state
= VlRecordState
.TERMINATED
2457 if vlr
.id in self
.vlr_uptime_tasks
:
2458 self
.vlr_uptime_tasks
[vlr
.id].cancel()
2460 self
._log
.debug("Terminating network service id %s", self
.id)
2462 # Move the state to TERMINATE
2463 self
.set_state(NetworkServiceRecordState
.TERMINATE
)
2464 event_descr
= "Terminate being processed for NS Id:%s" % self
.id
2465 self
.record_event("terminate", event_descr
)
2467 # Move the state to VNF_TERMINATE_PHASE
2468 self
._log
.debug("Terminating VNFFGs in NS ID: %s", self
.id)
2469 self
.set_state(NetworkServiceRecordState
.VNFFG_TERMINATE_PHASE
)
2470 event_descr
= "Terminating VNFFGS in NS Id:%s" % self
.id
2471 self
.record_event("terminating-vnffgss", event_descr
)
2472 yield from terminate_vnffgrs()
2474 # Move the state to VNF_TERMINATE_PHASE
2475 self
.set_state(NetworkServiceRecordState
.VNF_TERMINATE_PHASE
)
2476 event_descr
= "Terminating VNFS in NS Id:%s" % self
.id
2477 self
.record_event("terminating-vnfs", event_descr
)
2478 yield from self
.terminate_vnfrs(self
.vnfrs
.values())
2480 # Move the state to VL_TERMINATE_PHASE
2481 self
.set_state(NetworkServiceRecordState
.VL_TERMINATE_PHASE
)
2482 event_descr
= "Terminating VLs in NS Id:%s" % self
.id
2483 self
.record_event("terminating-vls", event_descr
)
2484 yield from terminate_vlrs()
2486 yield from self
.nsm_plugin
.terminate_ns(self
)
2488 # Move the state to TERMINATED
2489 self
.set_state(NetworkServiceRecordState
.TERMINATED
)
2490 event_descr
= "Terminated NS Id:%s" % self
.id
2491 self
.record_event("terminated", event_descr
)
2494 """"Enable a NetworkServiceRecord."""
2498 """"Disable a NetworkServiceRecord."""
2501 def map_config_status(self
):
2502 self
._log
.debug("Config status for ns {} is {}".
2503 format(self
.name
, self
._config
_status
))
2504 if self
._config
_status
== NsrYang
.ConfigStates
.CONFIGURING
:
2505 return 'configuring'
2506 if self
._config
_status
== NsrYang
.ConfigStates
.FAILED
:
2510 def vl_phase_completed(self
):
2511 """ Are VLs created in this NS?"""
2512 return self
._vl
_phase
_completed
2514 def vnf_phase_completed(self
):
2515 """ Are VLs created in this NS?"""
2516 return self
._vnf
_phase
_completed
2518 def create_msg(self
):
2519 """ The network serice record as a message """
2520 nsr_dict
= {"ns_instance_config_ref": self
.id}
2521 nsr
= RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr
.from_dict(nsr_dict
)
2522 #nsr.cloud_account = self.cloud_account_name
2523 nsr
.sdn_account
= self
._sdn
_account
_name
2524 nsr
.name_ref
= self
.name
2525 nsr
.nsd_ref
= self
.nsd_id
2526 nsr
.nsd_name_ref
= self
.nsd_msg
.name
2527 nsr
.operational_events
= self
._op
_status
.msg
2528 nsr
.operational_status
= self
._op
_status
.yang_str()
2529 nsr
.config_status
= self
.map_config_status()
2530 nsr
.config_status_details
= self
._config
_status
_details
2531 nsr
.create_time
= self
._create
_time
2532 nsr
.uptime
= int(time
.time()) - self
._create
_time
2534 for cfg_prim
in self
.nsd_msg
.service_primitive
:
2535 cfg_prim
= NsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr_ServicePrimitive
.from_dict(
2537 nsr
.service_primitive
.append(cfg_prim
)
2539 for init_cfg
in self
.nsd_msg
.initial_config_primitive
:
2540 prim
= NsrYang
.NsrInitialConfigPrimitive
.from_dict(
2542 nsr
.initial_config_primitive
.append(prim
)
2544 if self
.vl_phase_completed():
2545 for vlr
in self
.vlrs
:
2546 nsr
.vlr
.append(vlr
.create_nsr_vlr_msg(self
.vnfrs
.values()))
2548 if self
.vnf_phase_completed():
2549 for vnfr_id
in self
.vnfrs
:
2550 nsr
.constituent_vnfr_ref
.append(self
.vnfrs
[vnfr_id
].const_vnfr_msg
)
2551 for vnffgr
in self
.vnffgrs
.values():
2552 nsr
.vnffgr
.append(vnffgr
.fetch_vnffgr())
2553 for scaling_group
in self
._scaling
_groups
.values():
2554 nsr
.scaling_group_record
.append(scaling_group
.create_record_msg())
2558 def all_vnfs_active(self
):
2559 """ Are all VNFS in this NS active? """
2560 for _
, vnfr
in self
.vnfrs
.items():
2561 if vnfr
.active
is not True:
2566 def update_state(self
):
2567 """ Re-evaluate this NS's state """
2568 curr_state
= self
._op
_status
.state
2570 if curr_state
== NetworkServiceRecordState
.TERMINATED
:
2571 self
._log
.debug("NS (%s) in terminated state, not updating state", self
.id)
2574 new_state
= NetworkServiceRecordState
.RUNNING
2575 self
._log
.info("Received update_state for nsr: %s, curr-state: %s",
2576 self
.id, curr_state
)
2578 # Check all the VNFRs are present
2579 for _
, vnfr
in self
.vnfrs
.items():
2580 if vnfr
.state
in [VnfRecordState
.ACTIVE
, VnfRecordState
.TERMINATED
]:
2582 elif vnfr
.state
== VnfRecordState
.FAILED
:
2583 if vnfr
._prev
_state
!= vnfr
.state
:
2584 event_descr
= "Instantiation of VNF %s failed" % vnfr
.id
2585 event_error_details
= vnfr
.state_failed_reason
2586 self
.record_event("vnf-failed", event_descr
, evt_details
=event_error_details
)
2587 vnfr
.set_state(VnfRecordState
.FAILED
)
2589 self
._log
.info("VNF state did not change, curr=%s, prev=%s",
2590 vnfr
.state
, vnfr
._prev
_state
)
2591 new_state
= NetworkServiceRecordState
.FAILED
2594 self
._log
.info("VNF %s in NSR %s is still not active; current state is: %s",
2595 vnfr
.id, self
.id, vnfr
.state
)
2596 new_state
= curr_state
2598 # If new state is RUNNING; check all VLs
2599 if new_state
== NetworkServiceRecordState
.RUNNING
:
2600 for vl
in self
.vlrs
:
2602 if vl
.state
in [VlRecordState
.ACTIVE
, VlRecordState
.TERMINATED
]:
2604 elif vl
.state
== VlRecordState
.FAILED
:
2605 if vl
.prev_state
!= vl
.state
:
2606 event_descr
= "Instantiation of VL %s failed" % vl
.id
2607 event_error_details
= vl
.state_failed_reason
2608 self
.record_event("vl-failed", event_descr
, evt_details
=event_error_details
)
2609 vl
.prev_state
= vl
.state
2611 self
._log
.debug("VL %s already in failed state")
2613 if vl
.state
in [VlRecordState
.INSTANTIATION_PENDING
, VlRecordState
.INIT
]:
2614 new_state
= NetworkServiceRecordState
.VL_INSTANTIATE
2617 if vl
.state
in [VlRecordState
.TERMINATE_PENDING
]:
2618 new_state
= NetworkServiceRecordState
.VL_TERMINATE
2621 # If new state is RUNNING; check VNFFGRs are also active
2622 if new_state
== NetworkServiceRecordState
.RUNNING
:
2623 for _
, vnffgr
in self
.vnffgrs
.items():
2624 self
._log
.info("Checking vnffgr state for nsr %s is: %s",
2625 self
.id, vnffgr
.state
)
2626 if vnffgr
.state
== VnffgRecordState
.ACTIVE
:
2628 elif vnffgr
.state
== VnffgRecordState
.FAILED
:
2629 event_descr
= "Instantiation of VNFFGR %s failed" % vnffgr
.id
2630 self
.record_event("vnffg-failed", event_descr
)
2631 new_state
= NetworkServiceRecordState
.FAILED
2634 self
._log
.info("VNFFGR %s in NSR %s is still not active; current state is: %s",
2635 vnffgr
.id, self
.id, vnffgr
.state
)
2636 new_state
= curr_state
2638 # Update all the scaling group instance operational status to
2639 # reflect the state of all VNFR within that instance
2640 yield from self
._update
_scale
_group
_instances
_status
()
2642 for _
, group
in self
._scaling
_groups
.items():
2643 if group
.state
== scale_group
.ScaleGroupState
.SCALING_OUT
:
2644 new_state
= NetworkServiceRecordState
.SCALING_OUT
2646 elif group
.state
== scale_group
.ScaleGroupState
.SCALING_IN
:
2647 new_state
= NetworkServiceRecordState
.SCALING_IN
2650 if new_state
!= curr_state
:
2651 self
._log
.debug("Changing state of Network service %s from %s to %s",
2652 self
.id, curr_state
, new_state
)
2653 if new_state
== NetworkServiceRecordState
.RUNNING
:
2654 yield from self
.is_active()
2655 elif new_state
== NetworkServiceRecordState
.FAILED
:
2656 # If the NS is already active and we entered scaling_in, scaling_out,
2657 # do not mark the NS as failing if scaling operation failed.
2658 if curr_state
in [NetworkServiceRecordState
.SCALING_OUT
,
2659 NetworkServiceRecordState
.SCALING_IN
] and self
._is
_active
:
2660 new_state
= NetworkServiceRecordState
.RUNNING
2661 self
.set_state(new_state
)
2663 yield from self
.instantiation_failed()
2665 self
.set_state(new_state
)
2667 yield from self
.publish()
2670 class InputParameterSubstitution(object):
2672 This class is responsible for substituting input parameters into an NSD.
2675 def __init__(self
, log
, project
):
2676 """Create an instance of InputParameterSubstitution
2679 log - a logger for this object to use
2683 self
.project
= project
2685 def __call__(self
, nsd
, nsr_config
):
2686 """Substitutes input parameters from the NSR config into the NSD
2688 This call modifies the provided NSD with the input parameters that are
2689 contained in the NSR config.
2692 nsd - a GI NSD object
2693 nsr_config - a GI NSR config object
2696 if nsd
is None or nsr_config
is None:
2699 # Create a lookup of the xpath elements that this descriptor allows
2701 optional_input_parameters
= set()
2702 for input_parameter
in nsd
.input_parameter_xpath
:
2703 optional_input_parameters
.add(self
.project
.add_project(input_parameter
.xpath
))
2705 # Apply the input parameters to the descriptor
2706 if nsr_config
.input_parameter
:
2707 for param
in nsr_config
.input_parameter
:
2708 if param
.xpath
not in optional_input_parameters
:
2709 msg
= "tried to set an invalid input parameter ({})"
2710 self
.log
.error(msg
.format(param
.xpath
))
2714 "input-parameter:{} = {}".format(
2721 xpath
.setxattr(nsd
, param
.xpath
, param
.value
)
2723 except Exception as e
:
2724 self
.log
.exception(e
)
2727 class NetworkServiceDescriptor(object):
2729 Network service descriptor class
2732 def __init__(self
, dts
, log
, loop
, nsd
, nsm
):
2744 """ Returns nsd id """
2749 """ Returns name of nsd """
2750 return self
._nsd
.name
2753 def ref_count(self
):
2754 """ Returns reference count"""
2755 return self
._ref
_count
2758 """ Returns whether nsd is in use or not """
2759 return True if self
.ref_count
> 0 else False
2762 """ Take a reference on this object """
2763 self
._ref
_count
+= 1
2766 """ Release reference on this object """
2767 if self
.ref_count
< 1:
2768 msg
= ("Unref on a NSD object - nsd id %s, ref_count = %s" %
2769 (self
.id, self
.ref_count
))
2770 self
._log
.critical(msg
)
2771 raise NetworkServiceDescriptorError(msg
)
2772 self
._ref
_count
-= 1
2776 """ Return the message associated with this NetworkServiceDescriptor"""
2780 def path_for_id(nsd_id
):
2781 """ Return path for the passed nsd_id"""
2782 return self
._nsm
._project
.add_project(
2783 "C,/project-nsd:nsd-catalog/project-nsd:nsd[project-nsd:id = '{}'".
2787 """ Return the message associated with this NetworkServiceDescriptor"""
2788 return NetworkServiceDescriptor
.path_for_id(self
.id)
2790 def update(self
, nsd
):
2791 """ Update the NSD descriptor """
2795 class NsdDtsHandler(object):
2796 """ The network service descriptor DTS handler """
2797 XPATH
= "C,/project-nsd:nsd-catalog/project-nsd:nsd"
2799 def __init__(self
, dts
, log
, loop
, nsm
):
2806 self
._project
= nsm
._project
2810 """ Return registration handle """
2815 """ Register for Nsd create/update/delete/read requests from dts """
2818 self
._log
.warning("DTS handler already registered for project {}".
2819 format(self
._project
.name
))
2822 def on_apply(dts
, acg
, xact
, action
, scratch
):
2823 """Apply the configuration"""
2824 is_recovery
= xact
.xact
is None and action
== rwdts
.AppconfAction
.INSTALL
2825 self
._log
.debug("Got nsd apply cfg (xact:%s) (action:%s)",
2829 # Create/Update an NSD record
2830 for cfg
in self
._regh
.get_xact_elements(xact
):
2831 # Only interested in those NSD cfgs whose ID was received in prepare callback
2832 if cfg
.id in scratch
.get('nsds', []) or is_recovery
:
2833 self
._nsm
.update_nsd(cfg
)
2836 self
._log
.error("No reg handle for {} for project {}".
2837 format(self
.__class
__, self
._project
.name
))
2839 scratch
.pop('nsds', None)
2841 return RwTypes
.RwStatus
.SUCCESS
2844 def delete_nsd_libs(nsd_id
):
2845 """ Remove any files uploaded with NSD and stored under $RIFT_ARTIFACTS/libs/<id> """
2847 rift_artifacts_dir
= os
.environ
['RIFT_ARTIFACTS']
2848 nsd_dir
= os
.path
.join(rift_artifacts_dir
, 'launchpad/libs', nsd_id
)
2850 if os
.path
.exists (nsd_dir
):
2851 shutil
.rmtree(nsd_dir
, ignore_errors
=True)
2852 except Exception as e
:
2853 self
._log
.error("Exception in cleaning up NSD libs {}: {}".
2855 self
._log
.exception(e
)
2858 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2859 """ Prepare callback from DTS for NSD config """
2861 self
._log
.info("Got nsd prepare - config received nsd id %s, msg %s",
2864 fref
= ProtobufC
.FieldReference
.alloc()
2865 fref
.goto_whole_message(msg
.to_pbcm())
2867 if fref
.is_field_deleted():
2868 # Delete an NSD record
2869 self
._log
.debug("Deleting NSD with id %s", msg
.id)
2870 if self
._nsm
.nsd_in_use(msg
.id):
2871 self
._log
.debug("Cannot delete NSD in use - %s", msg
.id)
2872 err
= "Cannot delete an NSD in use - %s" % msg
.id
2873 raise NetworkServiceDescriptorRefCountExists(err
)
2875 yield from delete_nsd_libs(msg
.id)
2876 self
._nsm
.delete_nsd(msg
.id)
2878 # Add this NSD to scratch to create/update in apply callback
2879 nsds
= scratch
.setdefault('nsds', [])
2881 # acg._scratch['nsds'].append(msg.id)
2883 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2886 "Registering for NSD config using xpath: %s",
2887 NsdDtsHandler
.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 NSDs to create/update later
2893 # acg._scratch['nsds'] = list()
2894 self
._regh
= acg
.register(
2895 xpath
=self
._project
.add_project(NsdDtsHandler
.XPATH
),
2896 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
2897 on_prepare
=on_prepare
)
2899 def deregister(self
):
2900 self
._log
.debug("De-register NSD handler for project {}".
2901 format(self
._project
.name
))
2903 self
._regh
.deregister()
2907 class VnfdDtsHandler(object):
2908 """ DTS handler for VNFD config changes """
2909 XPATH
= "C,/project-vnfd:vnfd-catalog/project-vnfd:vnfd"
2911 def __init__(self
, dts
, log
, loop
, nsm
):
2917 self
._project
= nsm
._project
2921 """ DTS registration handle """
2926 """ Register for VNFD configuration"""
2929 self
._log
.warning("DTS handler already registered for project {}".
2930 format(self
._project
.name
))
2934 def on_apply(dts
, acg
, xact
, action
, scratch
):
2935 """Apply the configuration"""
2936 self
._log
.debug("Got NSM VNFD apply (xact: %s) (action: %s)(scr: %s)",
2937 xact
, action
, scratch
)
2940 # Create/Update a VNFD record
2941 for cfg
in self
._regh
.get_xact_elements(xact
):
2942 # Only interested in those VNFD cfgs whose ID was received in prepare callback
2943 if cfg
.id in scratch
.get('vnfds', []):
2944 self
._nsm
.update_vnfd(cfg
)
2946 for cfg
in self
._regh
.elements
:
2947 if cfg
.id in scratch
.get('deleted_vnfds', []):
2948 yield from self
._nsm
.delete_vnfd(cfg
.id)
2951 self
._log
.error("Reg handle none for {} in project {}".
2952 format(self
.__class
__, self
._project
))
2954 scratch
.pop('vnfds', None)
2955 scratch
.pop('deleted_vnfds', None)
2958 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2959 """ on prepare callback """
2960 self
._log
.debug("Got on prepare for VNFD (path: %s) (action: %s) (msg: %s)",
2961 ks_path
.to_xpath(RwNsmYang
.get_schema()), xact_info
.query_action
, msg
)
2963 fref
= ProtobufC
.FieldReference
.alloc()
2964 fref
.goto_whole_message(msg
.to_pbcm())
2966 # Handle deletes in prepare_callback, but adds/updates in apply_callback
2967 if fref
.is_field_deleted():
2968 self
._log
.debug("Adding msg to deleted field")
2969 deleted_vnfds
= scratch
.setdefault('deleted_vnfds', [])
2970 deleted_vnfds
.append(msg
.id)
2972 # Add this VNFD to scratch to create/update in apply callback
2973 vnfds
= scratch
.setdefault('vnfds', [])
2974 vnfds
.append(msg
.id)
2976 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2978 xpath
= self
._project
.add_project(VnfdDtsHandler
.XPATH
)
2980 "Registering for VNFD config using xpath {} for project {}"
2981 .format(xpath
, self
._project
))
2982 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2983 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2984 # Need a list in scratch to store VNFDs to create/update later
2985 # acg._scratch['vnfds'] = list()
2986 # acg._scratch['deleted_vnfds'] = list()
2987 self
._regh
= acg
.register(
2989 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY
,
2990 on_prepare
=on_prepare
)
2992 def deregister(self
):
2993 self
._log
.debug("De-register VNFD handler for project {}".
2994 format(self
._project
.name
))
2996 self
._regh
.deregister()
3000 class NsrRpcDtsHandler(object):
3001 """ The network service instantiation RPC DTS handler """
3002 EXEC_NSR_CONF_XPATH
= "I,/nsr:start-network-service"
3003 EXEC_NSR_CONF_O_XPATH
= "O,/nsr:start-network-service"
3004 NETCONF_IP_ADDRESS
= "127.0.0.1"
3006 RESTCONF_PORT
= 8888
3007 NETCONF_USER
= "admin"
3008 NETCONF_PW
= "admin"
3009 REST_BASE_V2_URL
= 'https://{}:{}/v2/api/'.format("127.0.0.1",8888)
3011 def __init__(self
, dts
, log
, loop
, nsm
):
3018 self
._ns
_regh
= None
3020 self
._manager
= None
3021 self
._nsr
_config
_url
= NsrRpcDtsHandler
.REST_BASE_V2_URL
+ \
3022 'config/project/{}/ns-instance-config'. \
3023 format(self
._nsm
._project
.name
)
3025 self
._model
= RwYang
.Model
.create_libncx()
3026 self
._model
.load_schema_ypbc(RwNsrYang
.get_schema())
3030 """ Return the NS manager instance """
3034 def wrap_netconf_config_xml(xml
):
3035 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
3039 def _connect(self
, timeout_secs
=240):
3041 start_time
= time
.time()
3042 while (time
.time() - start_time
) < timeout_secs
:
3045 self
._log
.debug("Attemping NsmTasklet netconf connection.")
3047 manager
= yield from ncclient
.asyncio_manager
.asyncio_connect(
3049 host
=NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
,
3050 port
=NsrRpcDtsHandler
.NETCONF_PORT
,
3051 username
=NsrRpcDtsHandler
.NETCONF_USER
,
3052 password
=NsrRpcDtsHandler
.NETCONF_PW
,
3054 look_for_keys
=False,
3055 hostkey_verify
=False,
3060 except ncclient
.transport
.errors
.SSHError
as e
:
3061 self
._log
.warning("Netconf connection to launchpad %s failed: %s",
3062 NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
, str(e
))
3064 yield from asyncio
.sleep(5, loop
=self
._loop
)
3066 raise NsrInstantiationFailed("Failed to connect to Launchpad within %s seconds" %
3069 def _apply_ns_instance_config(self
,payload_dict
):
3070 #self._log.debug("At apply NS instance config with payload %s",payload_dict)
3071 req_hdr
= {'accept':'application/vnd.yang.data+json',
3072 'content-type':'application/vnd.yang.data+json'}
3073 response
=requests
.post(self
._nsr
_config
_url
, headers
=req_hdr
,
3074 auth
=('admin', 'admin'),data
=payload_dict
,verify
=False)
3079 """ Register for NS monitoring read from dts """
3081 self
._log
.warning("RPC already registered for project {}".
3082 format(self
._project
.name
))
3086 def on_ns_config_prepare(xact_info
, action
, ks_path
, msg
):
3087 """ prepare callback from dts start-network-service"""
3088 assert action
== rwdts
.QueryAction
.RPC
3091 if not self
._nsm
._project
.rpc_check(msg
, xact_info
=xact_info
):
3094 rpc_op
= NsrYang
.YangOutput_Nsr_StartNetworkService
.from_dict({
3095 "nsr_id":str(uuid
.uuid4()),
3096 "project_name": msg
.prject_name
,
3099 if not ('name' in rpc_ip
and 'nsd_ref' in rpc_ip
and
3100 ('cloud_account' in rpc_ip
or 'om_datacenter' in rpc_ip
)):
3101 self
._log
.error("Mandatory parameters name or nsd_ref or " +
3102 "cloud account not found in start-network-service {}".
3106 self
._log
.debug("start-network-service RPC input: {}".format(rpc_ip
))
3109 # Add used value to the pool
3110 self
._log
.debug("RPC output: {}".format(rpc_op
))
3112 nsd_copy
= self
.nsm
.get_nsd(rpc_ip
.nsd_ref
)
3114 #if not self._manager:
3115 # self._manager = yield from self._connect()
3117 self
._log
.debug("Configuring ns-instance-config with name %s nsd-ref: %s",
3118 rpc_ip
.name
, rpc_ip
.nsd_ref
)
3120 ns_instance_config_dict
= {"id":rpc_op
.nsr_id
, "admin_status":"ENABLED"}
3121 ns_instance_config_copy_dict
= {k
:v
for k
, v
in rpc_ip
.as_dict().items()
3122 if k
in RwNsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr().fields
}
3123 ns_instance_config_dict
.update(ns_instance_config_copy_dict
)
3125 ns_instance_config
= RwNsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr
.from_dict(ns_instance_config_dict
)
3126 ns_instance_config
.nsd
= NsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr_Nsd()
3127 ns_instance_config
.nsd
.from_dict(nsd_copy
.msg
.as_dict())
3129 payload_dict
= ns_instance_config
.to_json(self
._model
)
3130 #xml = ns_instance_config.to_xml_v2(self._model)
3131 #netconf_xml = self.wrap_netconf_config_xml(xml)
3133 #self._log.debug("Sending configure ns-instance-config xml to %s: %s",
3134 # netconf_xml, NsrRpcDtsHandler.NETCONF_IP_ADDRESS)
3135 self
._log
.debug("Sending configure ns-instance-config json to %s: %s",
3136 self
._nsr
_config
_url
,ns_instance_config
)
3138 #response = yield from self._manager.edit_config(
3140 # config=netconf_xml,
3142 response
= yield from self
._loop
.run_in_executor(
3144 self
._apply
_ns
_instance
_config
,
3147 response
.raise_for_status()
3148 self
._log
.debug("Received edit config response: %s", response
.json())
3150 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
,
3151 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
,
3153 except Exception as e
:
3154 self
._log
.error("Exception processing the "
3155 "start-network-service: {}".format(e
))
3156 self
._log
.exception(e
)
3157 xact_info
.respond_xpath(rwdts
.XactRspCode
.NACK
,
3158 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
)
3161 hdl_ns
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_ns_config_prepare
,)
3163 with self
._dts
.group_create() as group
:
3164 self
._ns
_regh
= group
.register(xpath
=NsrRpcDtsHandler
.EXEC_NSR_CONF_XPATH
,
3166 flags
=rwdts
.Flag
.PUBLISHER
,
3169 def deregister(self
):
3170 self
._log
.debug("De-register NSR RPC for project {}".
3171 format(self
._nsm
._project
.name
))
3173 self
._ns
_regh
.deregister()
3174 self
._ns
_regh
= None
3177 class NsrDtsHandler(object):
3178 """ The network service DTS handler """
3179 NSR_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr"
3180 SCALE_INSTANCE_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr/nsr:scaling-group/nsr:instance"
3181 KEY_PAIR_XPATH
= "C,/nsr:key-pair"
3183 def __init__(self
, dts
, log
, loop
, nsm
):
3188 self
._project
= self
._nsm
._project
3190 self
._nsr
_regh
= None
3191 self
._scale
_regh
= None
3192 self
._key
_pair
_regh
= None
3196 """ Return the NS manager instance """
3201 """ Register for Nsr create/update/delete/read requests from dts """
3204 self
._log
.warning("DTS handler already registered for project {}".
3205 format(self
._project
.name
))
3208 def nsr_id_from_keyspec(ks
):
3209 nsr_path_entry
= NsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr
.schema().keyspec_to_entry(ks
)
3210 nsr_id
= nsr_path_entry
.key00
.id
3213 def group_name_from_keyspec(ks
):
3214 group_path_entry
= NsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup
.schema().keyspec_to_entry(ks
)
3215 group_name
= group_path_entry
.key00
.scaling_group_name_ref
3218 def is_instance_in_reg_elements(nsr_id
, group_name
, instance_id
):
3219 """ Return boolean indicating if scaling group instance was already commited previously.
3221 By looking at the existing elements in this registration handle (elements not part
3222 of this current xact), we can tell if the instance was configured previously without
3223 keeping any application state.
3225 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3226 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3227 elem_group_name
= group_name_from_keyspec(keyspec
)
3229 if elem_nsr_id
!= nsr_id
or group_name
!= elem_group_name
:
3232 if instance_cfg
.id == instance_id
:
3237 def get_scale_group_instance_delta(nsr_id
, group_name
, xact
):
3238 delta
= {"added": [], "deleted": []}
3239 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3240 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3241 if elem_nsr_id
!= nsr_id
:
3244 elem_group_name
= group_name_from_keyspec(keyspec
)
3245 if elem_group_name
!= group_name
:
3248 delta
["added"].append(instance_cfg
.id)
3250 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(include_keyspec
=True):
3251 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3252 if elem_nsr_id
!= nsr_id
:
3255 elem_group_name
= group_name_from_keyspec(keyspec
)
3256 if elem_group_name
!= group_name
:
3259 if instance_cfg
.id in delta
["added"]:
3260 delta
["added"].remove(instance_cfg
.id)
3262 delta
["deleted"].append(instance_cfg
.id)
3267 def update_nsr_nsd(nsr_id
, xact
, scratch
):
3270 def get_nsr_vl_delta(nsr_id
, xact
, scratch
):
3271 delta
= {"added": [], "deleted": []}
3272 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3273 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3274 if elem_nsr_id
!= nsr_id
:
3277 if 'vld' in instance_cfg
.nsd
:
3278 for vld
in instance_cfg
.nsd
.vld
:
3279 delta
["added"].append(vld
)
3281 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3282 self
._log
.debug("NSR update: %s", instance_cfg
)
3283 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3284 if elem_nsr_id
!= nsr_id
:
3287 if 'vld' in instance_cfg
.nsd
:
3288 for vld
in instance_cfg
.nsd
.vld
:
3289 if vld
in delta
["added"]:
3290 delta
["added"].remove(vld
)
3292 delta
["deleted"].append(vld
)
3296 vl_delta
= yield from get_nsr_vl_delta(nsr_id
, xact
, scratch
)
3297 self
._log
.debug("Got NSR:%s VL instance delta: %s", nsr_id
, vl_delta
)
3299 for vld
in vl_delta
["added"]:
3300 yield from self
._nsm
.nsr_instantiate_vl(nsr_id
, vld
)
3302 for vld
in vl_delta
["deleted"]:
3303 yield from self
._nsm
.nsr_terminate_vl(nsr_id
, vld
)
3305 def get_nsr_key_pairs(dts_member_reg
, xact
):
3307 for instance_cfg
, keyspec
in dts_member_reg
.get_xact_elements(xact
, include_keyspec
=True):
3308 self
._log
.debug("Key pair received is {} KS: {}".format(instance_cfg
, keyspec
))
3309 xpath
= keyspec
.to_xpath(RwNsrYang
.get_schema())
3310 key_pairs
[instance_cfg
.name
] = instance_cfg
3313 def on_apply(dts
, acg
, xact
, action
, scratch
):
3314 """Apply the configuration"""
3315 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3316 xact
, action
, scratch
)
3318 def handle_create_nsr(msg
, key_pairs
=None, restart_mode
=False):
3319 # Handle create nsr requests """
3320 # Do some validations
3321 if not msg
.has_field("nsd"):
3322 err
= "NSD not provided"
3323 self
._log
.error(err
)
3324 raise NetworkServiceRecordError(err
)
3326 self
._log
.debug("Creating NetworkServiceRecord %s from nsr config %s",
3327 msg
.id, msg
.as_dict())
3328 nsr
= self
.nsm
.create_nsr(msg
, key_pairs
=key_pairs
, restart_mode
=restart_mode
)
3331 def handle_delete_nsr(msg
):
3333 def delete_instantiation(ns_id
):
3334 """ Delete instantiation """
3335 with self
._dts
.transaction() as xact
:
3336 yield from self
._nsm
.terminate_ns(ns_id
, xact
)
3338 # Handle delete NSR requests
3339 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3340 # Terminate the NSR instance
3341 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3343 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3344 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3345 nsr
.record_event("terminate-rcvd", event_descr
)
3347 self
._loop
.create_task(delete_instantiation(msg
.id))
3350 def begin_instantiation(nsr
):
3351 # Begin instantiation
3352 self
._log
.info("Beginning NS instantiation: %s", nsr
.id)
3354 yield from self
._nsm
.instantiate_ns(nsr
.id, xact
)
3355 except Exception as e
:
3356 self
._log
.exception("NS instantiation: {}".format(e
))
3359 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3360 xact
, action
, scratch
)
3362 if action
== rwdts
.AppconfAction
.INSTALL
and xact
.id is None:
3364 if self
._key
_pair
_regh
:
3365 for element
in self
._key
_pair
_regh
.elements
:
3366 key_pairs
.append(element
)
3368 self
._log
.error("Reg handle none for key pair in project {}".
3369 format(self
._project
))
3372 for element
in self
._nsr
_regh
.elements
:
3373 nsr
= handle_create_nsr(element
, key_pairs
, restart_mode
=True)
3374 self
._loop
.create_task(begin_instantiation(nsr
))
3376 self
._log
.error("Reg handle none for NSR in project {}".
3377 format(self
._project
))
3380 (added_msgs
, deleted_msgs
, updated_msgs
) = get_add_delete_update_cfgs(self
._nsr
_regh
,
3383 self
._log
.debug("Added: %s, Deleted: %s, Updated: %s", added_msgs
,
3384 deleted_msgs
, updated_msgs
)
3386 for msg
in added_msgs
:
3387 if msg
.id not in self
._nsm
.nsrs
:
3388 self
._log
.info("Create NSR received in on_apply to instantiate NS:%s", msg
.id)
3389 key_pairs
= get_nsr_key_pairs(self
._key
_pair
_regh
, xact
)
3390 nsr
= handle_create_nsr(msg
,key_pairs
)
3391 self
._loop
.create_task(begin_instantiation(nsr
))
3393 for msg
in deleted_msgs
:
3394 self
._log
.info("Delete NSR received in on_apply to terminate NS:%s", msg
.id)
3396 handle_delete_nsr(msg
)
3398 self
._log
.exception("Failed to terminate NS:%s", msg
.id)
3400 for msg
in updated_msgs
:
3401 self
._log
.info("Update NSR received in on_apply: %s", msg
)
3403 self
._nsm
.nsr_update_cfg(msg
.id, msg
)
3406 self
._loop
.create_task(update_nsr_nsd(msg
.id, xact
, scratch
))
3408 for group
in msg
.scaling_group
:
3409 instance_delta
= get_scale_group_instance_delta(msg
.id, group
.scaling_group_name_ref
, xact
)
3410 self
._log
.debug("Got NSR:%s scale group instance delta: %s", msg
.id, instance_delta
)
3412 for instance_id
in instance_delta
["added"]:
3413 self
._nsm
.scale_nsr_out(msg
.id, group
.scaling_group_name_ref
, instance_id
, xact
)
3415 for instance_id
in instance_delta
["deleted"]:
3416 self
._nsm
.scale_nsr_in(msg
.id, group
.scaling_group_name_ref
, instance_id
)
3419 return RwTypes
.RwStatus
.SUCCESS
3422 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
3423 """ Prepare calllback from DTS for NSR """
3425 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3426 action
= xact_info
.query_action
3428 "Got Nsr prepare callback (xact: %s) (action: %s) (info: %s), %s:%s)",
3429 xact
, action
, xact_info
, xpath
, msg
3433 def delete_instantiation(ns_id
):
3434 """ Delete instantiation """
3435 yield from self
._nsm
.terminate_ns(ns_id
, None)
3437 def handle_delete_nsr():
3438 """ Handle delete NSR requests """
3439 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3440 # Terminate the NSR instance
3441 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3443 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3444 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3445 nsr
.record_event("terminate-rcvd", event_descr
)
3447 self
._loop
.create_task(delete_instantiation(msg
.id))
3449 fref
= ProtobufC
.FieldReference
.alloc()
3450 fref
.goto_whole_message(msg
.to_pbcm())
3452 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
, rwdts
.QueryAction
.DELETE
]:
3453 # if this is an NSR create
3454 if action
!= rwdts
.QueryAction
.DELETE
and msg
.id not in self
._nsm
.nsrs
:
3455 # Ensure the Cloud account/datacenter has been specified
3456 if not msg
.has_field("cloud_account") and not msg
.has_field("om_datacenter"):
3457 raise NsrInstantiationFailed("Cloud account or datacenter not specified in NSR")
3459 # Check if nsd is specified
3460 if not msg
.has_field("nsd"):
3461 raise NsrInstantiationFailed("NSD not specified in NSR")
3464 nsr
= self
._nsm
.nsrs
[msg
.id]
3466 if msg
.has_field("nsd"):
3467 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3468 raise NsrVlUpdateError("Unable to update VL when NSR not in running state")
3469 if 'vld' not in msg
.nsd
or len(msg
.nsd
.vld
) == 0:
3470 raise NsrVlUpdateError("NS config NSD should have atleast 1 VLD defined")
3472 if msg
.has_field("scaling_group"):
3473 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3474 raise ScalingOperationError("Unable to perform scaling action when NS is not in running state")
3476 if len(msg
.scaling_group
) > 1:
3477 raise ScalingOperationError("Only a single scaling group can be configured at a time")
3479 for group_msg
in msg
.scaling_group
:
3480 num_new_group_instances
= len(group_msg
.instance
)
3481 if num_new_group_instances
> 1:
3482 raise ScalingOperationError("Only a single scaling instance can be modified at a time")
3484 elif num_new_group_instances
== 1:
3485 scale_group
= nsr
.scaling_groups
[group_msg
.scaling_group_name_ref
]
3486 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
]:
3487 if len(scale_group
.instances
) == scale_group
.max_instance_count
:
3488 raise ScalingOperationError("Max instances for %s reached" % scale_group
)
3490 acg
.handle
.prepare_complete_ok(xact_info
.handle
)
3493 xpath
= self
._project
.add_project(NsrDtsHandler
.NSR_XPATH
)
3494 self
._log
.debug("Registering for NSR config using xpath: {}".
3497 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
3498 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
3499 self
._nsr
_regh
= acg
.register(
3501 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3502 on_prepare
=on_prepare
3505 self
._scale
_regh
= acg
.register(
3506 xpath
=self
._project
.add_project(NsrDtsHandler
.SCALE_INSTANCE_XPATH
),
3507 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY| rwdts
.Flag
.CACHE
,
3510 self
._key
_pair
_regh
= acg
.register(
3511 xpath
=self
._project
.add_project(NsrDtsHandler
.KEY_PAIR_XPATH
),
3512 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3515 def deregister(self
):
3516 self
._log
.debug("De-register NSR config for project {}".
3517 format(self
._project
.name
))
3519 self
._nsr
_regh
.deregister()
3520 self
._nsr
_regh
= None
3521 if self
._scale
_regh
:
3522 self
._scale
_regh
.deregister()
3523 self
._scale
_regh
= None
3524 if self
._key
_pair
_regh
:
3525 self
._key
_pair
_regh
.deregister()
3526 self
._key
_pair
_regh
= None
3529 class NsrOpDataDtsHandler(object):
3530 """ The network service op data DTS handler """
3531 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
3533 def __init__(self
, dts
, log
, loop
, nsm
):
3539 self
._project
= nsm
._project
3544 """ Return the registration handle"""
3549 """ Return the NS manager instance """
3554 """ Register for Nsr op data publisher registration"""
3556 self
._log
.warning("NSR op data handler already registered for project {}".
3557 format(self
._project
.name
))
3560 xpath
= self
._project
.add_project(NsrOpDataDtsHandler
.XPATH
)
3561 self
._log
.debug("Registering Nsr op data path {} as publisher".
3564 hdl
= rift
.tasklets
.DTS
.RegistrationHandler()
3565 handlers
= rift
.tasklets
.Group
.Handler()
3566 with self
._dts
.group_create(handler
=handlers
) as group
:
3567 self
._regh
= group
.register(xpath
=xpath
,
3569 flags
=rwdts
.Flag
.PUBLISHER | rwdts
.Flag
.NO_PREP_READ | rwdts
.Flag
.DATASTORE
)
3571 def deregister(self
):
3572 self
._log
.debug("De-register NSR opdata for project {}".
3573 format(self
._project
.name
))
3575 self
._regh
.deregister()
3579 def create(self
, xpath
, msg
):
3581 Create an NS record in DTS with the path and message
3583 path
= self
._project
.add_project(xpath
)
3584 self
._log
.debug("Creating NSR %s:%s", path
, msg
)
3585 self
.regh
.create_element(path
, msg
)
3586 self
._log
.debug("Created NSR, %s:%s", path
, msg
)
3589 def update(self
, xpath
, msg
, flags
=rwdts
.XactFlag
.REPLACE
):
3591 Update an NS record in DTS with the path and message
3593 path
= self
._project
.add_project(xpath
)
3594 self
._log
.debug("Updating NSR, %s:%s regh = %s", path
, msg
, self
.regh
)
3595 self
.regh
.update_element(path
, msg
, flags
)
3596 self
._log
.debug("Updated NSR, %s:%s", path
, msg
)
3599 def delete(self
, xpath
):
3601 Update an NS record in DTS with the path and message
3603 path
= self
._project
.add_project(xpath
)
3604 self
._log
.debug("Deleting NSR path:%s", path
)
3605 self
.regh
.delete_element(path
)
3606 self
._log
.debug("Deleted NSR path:%s", path
)
3609 class VnfrDtsHandler(object):
3610 """ The virtual network service DTS handler """
3611 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
3613 def __init__(self
, dts
, log
, loop
, nsm
):
3623 """ Return registration handle """
3628 """ Return the NS manager instance """
3633 """ Register for vnfr create/update/delete/ advises from dts """
3635 self
._log
.warning("VNFR DTS handler already registered for project {}".
3636 format(self
._project
.name
))
3640 def on_commit(xact_info
):
3641 """ The transaction has been committed """
3642 self
._log
.debug("Got vnfr commit (xact_info: %s)", xact_info
)
3643 return rwdts
.MemberRspCode
.ACTION_OK
3646 def on_prepare(xact_info
, action
, ks_path
, msg
):
3647 """ prepare callback from dts """
3648 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3650 "Got vnfr on_prepare cb (xact_info: %s, action: %s): %s:%s",
3651 xact_info
, action
, ks_path
, msg
3654 schema
= VnfrYang
.YangData_RwProject_Project_VnfrCatalog_Vnfr
.schema()
3655 path_entry
= schema
.keyspec_to_entry(ks_path
)
3656 if not path_entry
or (path_entry
.key00
.id not in self
._nsm
._vnfrs
):
3657 # Check if this is a monitoring param xpath
3658 if 'vnfr:monitoring-param' not in xpath
:
3659 self
._log
.error("%s request for non existent record path %s",
3661 xact_info
.respond_xpath(rwdts
.XactRspCode
.NA
, xpath
)
3665 if action
== rwdts
.QueryAction
.CREATE
or action
== rwdts
.QueryAction
.UPDATE
:
3666 yield from self
._nsm
.update_vnfr(msg
)
3667 elif action
== rwdts
.QueryAction
.DELETE
:
3668 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3669 self
._nsm
.delete_vnfr(path_entry
.key00
.id)
3671 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
, xpath
)
3673 self
._log
.debug("Registering for VNFR using xpath: %s",
3674 VnfrDtsHandler
.XPATH
,)
3676 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_commit
=on_commit
,
3677 on_prepare
=on_prepare
,)
3678 with self
._dts
.group_create() as group
:
3679 self
._regh
= group
.register(xpath
=self
._nsm
._project
.add_project(
3680 VnfrDtsHandler
.XPATH
),
3682 flags
=(rwdts
.Flag
.SUBSCRIBER
),)
3684 def deregister(self
):
3685 self
._log
.debug("De-register VNFR for project {}".
3686 format(self
._nsm
._project
.name
))
3688 self
._regh
.deregister()
3691 class NsdRefCountDtsHandler(object):
3692 """ The NSD Ref Count DTS handler """
3693 XPATH
= "D,/nsr:ns-instance-opdata/rw-nsr:nsd-ref-count"
3695 def __init__(self
, dts
, log
, loop
, nsm
):
3705 """ Return registration handle """
3710 """ Return the NS manager instance """
3715 """ Register for NSD ref count read from dts """
3717 self
._log
.warning("NSD ref DTS handler already registered for project {}".
3718 format(self
._project
.name
))
3723 def on_prepare(xact_info
, action
, ks_path
, msg
):
3724 """ prepare callback from dts """
3725 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3727 if action
== rwdts
.QueryAction
.READ
:
3728 schema
= RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount
.schema()
3729 path_entry
= schema
.keyspec_to_entry(ks_path
)
3730 nsd_list
= yield from self
._nsm
.get_nsd_refcount(path_entry
.key00
.nsd_id_ref
)
3731 for xpath
, msg
in nsd_list
:
3732 xact_info
.respond_xpath(rsp_code
=rwdts
.XactRspCode
.MORE
,
3735 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
3737 raise NetworkServiceRecordError("Not supported operation %s" % action
)
3739 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_prepare
,)
3740 with self
._dts
.group_create() as group
:
3741 self
._regh
= group
.register(xpath
=self
._nsm
._project
.add_project(
3742 NsdRefCountDtsHandler
.XPATH
),
3744 flags
=rwdts
.Flag
.PUBLISHER
,)
3746 def deregister(self
):
3747 self
._log
.debug("De-register NSD Ref count for project {}".
3748 format(self
._nsm
._project
.name
))
3750 self
._regh
.deregister()
3754 class NsManager(object):
3755 """ The Network Service Manager class"""
3756 def __init__(self
, dts
, log
, loop
, project
,
3757 nsr_handler
, vnfr_handler
, vlr_handler
, ro_plugin_selector
,
3758 vnffgmgr
, vnfd_pub_handler
, cloud_account_handler
):
3762 self
._project
= project
3763 self
._nsr
_handler
= nsr_handler
3764 self
._vnfr
_pub
_handler
= vnfr_handler
3765 self
._vlr
_pub
_handler
= vlr_handler
3766 self
._vnffgmgr
= vnffgmgr
3767 self
._vnfd
_pub
_handler
= vnfd_pub_handler
3768 self
._cloud
_account
_handler
= cloud_account_handler
3770 self
._ro
_plugin
_selector
= ro_plugin_selector
3771 self
._ncclient
= rift
.mano
.ncclient
.NcClient(
3783 self
.cfgmgr_obj
= conman
.ROConfigManager(log
, loop
, dts
, self
)
3785 # TODO: All these handlers should move to tasklet level.
3786 # Passing self is often an indication of bad design
3787 self
._nsd
_dts
_handler
= NsdDtsHandler(dts
, log
, loop
, self
)
3788 self
._vnfd
_dts
_handler
= VnfdDtsHandler(dts
, log
, loop
, self
)
3789 self
._dts
_handlers
= [self
._nsd
_dts
_handler
,
3790 VnfrDtsHandler(dts
, log
, loop
, self
),
3791 NsdRefCountDtsHandler(dts
, log
, loop
, self
),
3792 NsrDtsHandler(dts
, log
, loop
, self
),
3793 ScalingRpcHandler(log
, dts
, loop
, self
._project
,
3794 self
.scale_rpc_callback
),
3795 NsrRpcDtsHandler(dts
, log
, loop
, self
),
3796 self
._vnfd
_dts
_handler
,
3817 def nsr_handler(self
):
3818 """" NSR handler """
3819 return self
._nsr
_handler
3823 """" So Obj handler """
3828 """ NSRs in this NSM"""
3833 """ NSDs in this NSM"""
3838 """ VNFDs in this NSM"""
3843 """ VNFRs in this NSM"""
3847 def nsr_pub_handler(self
):
3848 """ NSR publication handler """
3849 return self
._nsr
_handler
3852 def vnfr_pub_handler(self
):
3853 """ VNFR publication handler """
3854 return self
._vnfr
_pub
_handler
3857 def vlr_pub_handler(self
):
3858 """ VLR publication handler """
3859 return self
._vlr
_pub
_handler
3862 def vnfd_pub_handler(self
):
3863 return self
._vnfd
_pub
_handler
3867 """ Register all static DTS handlers """
3868 for dts_handle
in self
._dts
_handlers
:
3869 yield from dts_handle
.register()
3871 def deregister(self
):
3872 """ Register all static DTS handlers """
3873 for dts_handle
in self
._dts
_handlers
:
3874 dts_handle
.deregister()
3877 def get_ns_by_nsr_id(self
, nsr_id
):
3878 """ get NSR by nsr id """
3879 if nsr_id
not in self
._nsrs
:
3880 raise NetworkServiceRecordError("NSR id %s not found" % nsr_id
)
3882 return self
._nsrs
[nsr_id
]
3884 def scale_nsr_out(self
, nsr_id
, scale_group_name
, instance_id
, config_xact
):
3885 self
.log
.debug("Scale out NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3890 nsr
= self
._nsrs
[nsr_id
]
3891 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3892 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3894 self
._loop
.create_task(nsr
.create_scale_group_instance(scale_group_name
, instance_id
, config_xact
))
3896 def scale_nsr_in(self
, nsr_id
, scale_group_name
, instance_id
):
3897 self
.log
.debug("Scale in NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3902 nsr
= self
._nsrs
[nsr_id
]
3903 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3904 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3906 self
._loop
.create_task(nsr
.delete_scale_group_instance(scale_group_name
, instance_id
))
3908 def scale_rpc_callback(self
, xact
, msg
, action
):
3909 """Callback handler for RPC calls
3911 xact : Transaction Handler
3913 action : Scaling Action
3915 ScalingGroupInstance
= NsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup_Instance
3916 ScalingGroup
= NsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup
3918 xpath
= self
._project
.add_project(
3919 ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').
3920 format(msg
.nsr_id_ref
))
3922 instance
= ScalingGroupInstance
.from_dict({
3923 "id": msg
.instance_id
,
3924 "project_name": self
._project
.name
,})
3927 def get_nsr_scaling_group():
3928 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3930 for result
in results
:
3931 res
= yield from result
3932 nsr_config
= res
.result
3934 for scaling_group
in nsr_config
.scaling_group
:
3935 if scaling_group
.scaling_group_name_ref
== msg
.scaling_group_name_ref
:
3938 scaling_group
= nsr_config
.scaling_group
.add()
3939 scaling_group
.scaling_group_name_ref
= msg
.scaling_group_name_ref
3941 return (nsr_config
, scaling_group
)
3944 def update_config(nsr_config
):
3945 xml
= self
._ncclient
.convert_to_xml(RwNsrYang
, nsr_config
)
3946 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
3947 yield from self
._ncclient
.connect()
3948 yield from self
._ncclient
.manager
.edit_config(target
="running", config
=xml
, default_operation
="replace")
3952 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3953 scaling_group
.instance
.append(instance
)
3954 yield from update_config(nsr_config
)
3958 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3959 scaling_group
.instance
.remove(instance
)
3960 yield from update_config(nsr_config
)
3962 if action
== ScalingRpcHandler
.ACTION
.SCALE_OUT
:
3963 self
._loop
.create_task(scale_out())
3965 self
._loop
.create_task(scale_in())
3967 # Opdata based calls, disabled for now!
3968 # if action == ScalingRpcHandler.ACTION.SCALE_OUT:
3969 # self.scale_nsr_out(
3971 # msg.scaling_group_name_ref,
3975 # self.scale_nsr_in(
3977 # msg.scaling_group_name_ref,
3980 def nsr_update_cfg(self
, nsr_id
, msg
):
3981 nsr
= self
._nsrs
[nsr_id
]
3982 nsr
.nsr_cfg_msg
= msg
3984 def nsr_instantiate_vl(self
, nsr_id
, vld
):
3985 self
.log
.debug("NSR {} create VL {}".format(nsr_id
, vld
))
3986 nsr
= self
._nsrs
[nsr_id
]
3987 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3988 raise NsrVlUpdateError("Cannot perform VL instantiate if NSR is not in running state")
3990 # Not calling in a separate task as this is called from a separate task
3991 yield from nsr
.create_vl_instance(vld
)
3993 def nsr_terminate_vl(self
, nsr_id
, vld
):
3994 self
.log
.debug("NSR {} delete VL {}".format(nsr_id
, vld
.id))
3995 nsr
= self
._nsrs
[nsr_id
]
3996 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3997 raise NsrVlUpdateError("Cannot perform VL terminate if NSR is not in running state")
3999 # Not calling in a separate task as this is called from a separate task
4000 yield from nsr
.delete_vl_instance(vld
)
4002 def create_nsr(self
, nsr_msg
, key_pairs
=None,restart_mode
=False):
4003 """ Create an NSR instance """
4004 if nsr_msg
.id in self
._nsrs
:
4005 msg
= "NSR id %s already exists" % nsr_msg
.id
4006 self
._log
.error(msg
)
4007 raise NetworkServiceRecordError(msg
)
4009 self
._log
.info("Create NetworkServiceRecord nsr id %s from nsd_id %s, restart mode %s",
4014 nsm_plugin
= self
._ro
_plugin
_selector
.ro_plugin
4015 sdn_account_name
= self
._cloud
_account
_handler
.get_cloud_account_sdn_name(nsr_msg
.cloud_account
)
4017 nsr
= NetworkServiceRecord(self
._dts
,
4026 restart_mode
=restart_mode
,
4027 vlr_handler
=self
._ro
_plugin
_selector
._records
_publisher
._vlr
_pub
_hdlr
4029 self
._nsrs
[nsr_msg
.id] = nsr
4030 nsm_plugin
.create_nsr(nsr_msg
, nsr_msg
.nsd
, key_pairs
)
4034 def delete_nsr(self
, nsr_id
):
4036 Delete NSR with the passed nsr id
4038 del self
._nsrs
[nsr_id
]
4041 def instantiate_ns(self
, nsr_id
, config_xact
):
4042 """ Instantiate an NS instance """
4043 self
._log
.debug("Instantiating Network service id %s", nsr_id
)
4044 if nsr_id
not in self
._nsrs
:
4045 err
= "NSR id %s not found " % nsr_id
4046 self
._log
.error(err
)
4047 raise NetworkServiceRecordError(err
)
4049 nsr
= self
._nsrs
[nsr_id
]
4050 yield from nsr
.nsm_plugin
.instantiate_ns(nsr
, config_xact
)
4053 def update_vnfr(self
, vnfr
):
4054 """Create/Update an VNFR """
4056 vnfr_state
= self
._vnfrs
[vnfr
.id].state
4057 self
._log
.debug("Updating VNFR with state %s: vnfr %s", vnfr_state
, vnfr
)
4059 yield from self
._vnfrs
[vnfr
.id].update_state(vnfr
)
4060 nsr
= self
.find_nsr_for_vnfr(vnfr
.id)
4061 yield from nsr
.update_state()
4063 def find_nsr_for_vnfr(self
, vnfr_id
):
4064 """ Find the NSR which )has the passed vnfr id"""
4065 for nsr
in list(self
.nsrs
.values()):
4066 for vnfr
in list(nsr
.vnfrs
.values()):
4067 if vnfr
.id == vnfr_id
:
4071 def delete_vnfr(self
, vnfr_id
):
4072 """ Delete VNFR with the passed id"""
4073 del self
._vnfrs
[vnfr_id
]
4075 def get_nsd_ref(self
, nsd_id
):
4076 """ Get network service descriptor for the passed nsd_id
4078 nsd
= self
.get_nsd(nsd_id
)
4083 def get_nsr_config(self
, nsd_id
):
4084 xpath
= self
._project
.add_project("C,/nsr:ns-instance-config")
4085 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
4087 for result
in results
:
4088 entry
= yield from result
4089 ns_instance_config
= entry
.result
4091 for nsr
in ns_instance_config
.nsr
:
4092 if nsr
.nsd
.id == nsd_id
:
4098 def nsd_unref_by_nsr_id(self
, nsr_id
):
4099 """ Unref the network service descriptor based on NSR id """
4100 self
._log
.debug("NSR Unref called for Nsr Id:%s", nsr_id
)
4101 if nsr_id
in self
._nsrs
:
4102 nsr
= self
._nsrs
[nsr_id
]
4105 nsd
= self
.get_nsd(nsr
.nsd_id
)
4106 self
._log
.debug("Releasing ref on NSD %s held by NSR %s - Curr %d",
4107 nsd
.id, nsr
.id, nsd
.ref_count
)
4109 except NetworkServiceDescriptorError
:
4110 # We store a copy of NSD in NSR and the NSD in nsd-catalog
4115 self
._log
.error("Cannot find NSR with id %s", nsr_id
)
4116 raise NetworkServiceDescriptorUnrefError("No NSR with id" % nsr_id
)
4119 def nsd_unref(self
, nsd_id
):
4120 """ Unref the network service descriptor associated with the id """
4121 nsd
= self
.get_nsd(nsd_id
)
4124 def get_nsd(self
, nsd_id
):
4125 """ Get network service descriptor for the passed nsd_id"""
4126 if nsd_id
not in self
._nsds
:
4127 self
._log
.error("Cannot find NSD id:%s", nsd_id
)
4128 raise NetworkServiceDescriptorError("Cannot find NSD id:%s", nsd_id
)
4130 return self
._nsds
[nsd_id
]
4132 def create_nsd(self
, nsd_msg
):
4133 """ Create a network service descriptor """
4134 self
._log
.debug("Create network service descriptor - %s", nsd_msg
)
4135 if nsd_msg
.id in self
._nsds
:
4136 self
._log
.error("Cannot create NSD %s -NSD ID already exists", nsd_msg
)
4137 raise NetworkServiceDescriptorError("NSD already exists-%s", nsd_msg
.id)
4139 nsd
= NetworkServiceDescriptor(
4146 self
._nsds
[nsd_msg
.id] = nsd
4150 def update_nsd(self
, nsd
):
4151 """ update the Network service descriptor """
4152 self
._log
.debug("Update network service descriptor - %s", nsd
)
4153 if nsd
.id not in self
._nsds
:
4154 self
._log
.debug("No NSD found - creating NSD id = %s", nsd
.id)
4155 self
.create_nsd(nsd
)
4157 self
._log
.debug("Updating NSD id = %s, nsd = %s", nsd
.id, nsd
)
4158 self
._nsds
[nsd
.id].update(nsd
)
4160 def delete_nsd(self
, nsd_id
):
4161 """ Delete the Network service descriptor with the passed id """
4162 self
._log
.debug("Deleting the network service descriptor - %s", nsd_id
)
4163 if nsd_id
not in self
._nsds
:
4164 self
._log
.debug("Delete NSD failed - cannot find nsd-id %s", nsd_id
)
4165 raise NetworkServiceDescriptorNotFound("Cannot find %s", nsd_id
)
4167 if nsd_id
not in self
._nsds
:
4168 self
._log
.debug("Cannot delete NSD id %s reference exists %s",
4170 self
._nsds
[nsd_id
].ref_count
)
4171 raise NetworkServiceDescriptorRefCountExists(
4172 "Cannot delete :%s, ref_count:%s",
4174 self
._nsds
[nsd_id
].ref_count
)
4176 del self
._nsds
[nsd_id
]
4178 def get_vnfd_config(self
, xact
):
4179 vnfd_dts_reg
= self
._vnfd
_dts
_handler
.regh
4180 for cfg
in vnfd_dts_reg
.get_xact_elements(xact
):
4181 if cfg
.id not in self
._vnfds
:
4182 self
.create_vnfd(cfg
)
4184 def get_vnfd(self
, vnfd_id
, xact
):
4185 """ Get virtual network function descriptor for the passed vnfd_id"""
4186 if vnfd_id
not in self
._vnfds
:
4187 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4188 self
.get_vnfd_config(xact
)
4190 if vnfd_id
not in self
._vnfds
:
4191 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4192 raise VnfDescriptorError("Cannot find VNFD id:%s", vnfd_id
)
4194 return self
._vnfds
[vnfd_id
]
4196 def create_vnfd(self
, vnfd
):
4197 """ Create a virtual network function descriptor """
4198 self
._log
.debug("Create virtual network function descriptor - %s", vnfd
)
4199 if vnfd
.id in self
._vnfds
:
4200 self
._log
.error("Cannot create VNFD %s -VNFD ID already exists", vnfd
)
4201 raise VnfDescriptorError("VNFD already exists-%s", vnfd
.id)
4203 self
._vnfds
[vnfd
.id] = vnfd
4204 return self
._vnfds
[vnfd
.id]
4206 def update_vnfd(self
, vnfd
):
4207 """ Update the virtual network function descriptor """
4208 self
._log
.debug("Update virtual network function descriptor- %s", vnfd
)
4211 if vnfd
.id not in self
._vnfds
:
4212 self
._log
.debug("No VNFD found - creating VNFD id = %s", vnfd
.id)
4213 self
.create_vnfd(vnfd
)
4215 self
._log
.debug("Updating VNFD id = %s, vnfd = %s", vnfd
.id, vnfd
)
4216 self
._vnfds
[vnfd
.id] = vnfd
4219 def delete_vnfd(self
, vnfd_id
):
4220 """ Delete the virtual network function descriptor with the passed id """
4221 self
._log
.debug("Deleting the virtual network function descriptor - %s", vnfd_id
)
4222 if vnfd_id
not in self
._vnfds
:
4223 self
._log
.debug("Delete VNFD failed - cannot find vnfd-id %s", vnfd_id
)
4224 raise VnfDescriptorError("Cannot find %s", vnfd_id
)
4226 del self
._vnfds
[vnfd_id
]
4228 def nsd_in_use(self
, nsd_id
):
4229 """ Is the NSD with the passed id in use """
4230 self
._log
.debug("Is this NSD in use - msg:%s", nsd_id
)
4231 if nsd_id
in self
._nsds
:
4232 return self
._nsds
[nsd_id
].in_use()
4236 def publish_nsr(self
, xact
, path
, msg
):
4237 """ Publish a NSR """
4238 self
._log
.debug("Publish NSR with path %s, msg %s",
4240 yield from self
.nsr_handler
.update(xact
, path
, msg
)
4243 def unpublish_nsr(self
, xact
, path
):
4244 """ Un Publish an NSR """
4245 self
._log
.debug("Publishing delete NSR with path %s", path
)
4246 yield from self
.nsr_handler
.delete(path
, xact
)
4248 def vnfr_is_ready(self
, vnfr_id
):
4249 """ VNFR with the id is ready """
4250 self
._log
.debug("VNFR id %s ready", vnfr_id
)
4251 if vnfr_id
not in self
._vnfds
:
4252 err
= "Did not find VNFR ID with id %s" % vnfr_id
4253 self
._log
.critical("err")
4254 raise VirtualNetworkFunctionRecordError(err
)
4255 self
._vnfrs
[vnfr_id
].is_ready()
4258 def get_nsd_refcount(self
, nsd_id
):
4259 """ Get the nsd_list from this NSM"""
4261 def nsd_refcount_xpath(nsd_id
):
4262 """ xpath for ref count entry """
4263 return (self
._project
.add_project(NsdRefCountDtsHandler
.XPATH
) +
4264 "[rw-nsr:nsd-id-ref = '{}']").format(nsd_id
)
4267 if nsd_id
is None or nsd_id
== "":
4268 for nsd
in self
._nsds
.values():
4269 nsd_msg
= RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount()
4270 nsd_msg
.nsd_id_ref
= nsd
.id
4271 nsd_msg
.instance_ref_count
= nsd
.ref_count
4272 nsd_list
.append((nsd_refcount_xpath(nsd
.id), nsd_msg
))
4273 elif nsd_id
in self
._nsds
:
4274 nsd_msg
= RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount()
4275 nsd_msg
.nsd_id_ref
= self
._nsds
[nsd_id
].id
4276 nsd_msg
.instance_ref_count
= self
._nsds
[nsd_id
].ref_count
4277 nsd_list
.append((nsd_refcount_xpath(nsd_id
), nsd_msg
))
4282 def terminate_ns(self
, nsr_id
, xact
):
4284 Terminate network service for the given NSR Id
4287 # Terminate the instances/networks assocaited with this nw service
4288 self
._log
.debug("Terminating the network service %s", nsr_id
)
4290 yield from self
._nsrs
[nsr_id
].terminate()
4291 except Exception as e
:
4292 self
.log
.exception("Failed to terminate NSR[id=%s]", nsr_id
)
4295 yield from self
.nsd_unref_by_nsr_id(nsr_id
)
4297 # Unpublish the NSR record
4298 self
._log
.debug("Unpublishing the network service %s", nsr_id
)
4299 yield from self
._nsrs
[nsr_id
].unpublish(xact
)
4301 # Finaly delete the NS instance from this NS Manager
4302 self
._log
.debug("Deletng the network service %s", nsr_id
)
4303 self
.delete_nsr(nsr_id
)
4306 class NsmRecordsPublisherProxy(object):
4307 """ This class provides a publisher interface that allows plugin objects
4308 to publish NSR/VNFR/VLR"""
4310 def __init__(self
, dts
, log
, loop
, project
, nsr_pub_hdlr
,
4311 vnfr_pub_hdlr
, vlr_pub_hdlr
,):
4315 self
._project
= project
4316 self
._nsr
_pub
_hdlr
= nsr_pub_hdlr
4317 self
._vlr
_pub
_hdlr
= vlr_pub_hdlr
4318 self
._vnfr
_pub
_hdlr
= vnfr_pub_hdlr
4321 def publish_nsr(self
, xact
, nsr
):
4322 """ Publish an NSR """
4323 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4324 return (yield from self
._nsr
_pub
_hdlr
.update(xact
, path
, nsr
))
4327 def unpublish_nsr(self
, xact
, nsr
):
4328 """ Unpublish an NSR """
4329 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4330 return (yield from self
._nsr
_pub
_hdlr
.delete(xact
, path
))
4333 def publish_vnfr(self
, xact
, vnfr
):
4334 """ Publish an VNFR """
4335 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4336 return (yield from self
._vnfr
_pub
_hdlr
.update(xact
, path
, vnfr
))
4339 def unpublish_vnfr(self
, xact
, vnfr
):
4340 """ Unpublish a VNFR """
4341 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4342 return (yield from self
._vnfr
_pub
_hdlr
.delete(xact
, path
))
4345 def publish_vlr(self
, xact
, vlr
):
4346 """ Publish a VLR """
4347 path
= self
._project
.add_project(VirtualLinkRecord
.vlr_xpath(vlr
))
4348 return (yield from self
._vlr
_pub
_hdlr
.update(xact
, path
, vlr
))
4351 def unpublish_vlr(self
, xact
, vlr
):
4352 """ Unpublish a VLR """
4353 path
= self
._project
.add_project(VirtualLinkRecord
.vlr_xpath(vlr
))
4354 return (yield from self
._vlr
_pub
_hdlr
.delete(xact
, path
))
4357 class ScalingRpcHandler(mano_dts
.DtsHandler
):
4358 """ The Network service Monitor DTS handler """
4359 SCALE_IN_INPUT_XPATH
= "I,/nsr:exec-scale-in"
4360 SCALE_IN_OUTPUT_XPATH
= "O,/nsr:exec-scale-in"
4362 SCALE_OUT_INPUT_XPATH
= "I,/nsr:exec-scale-out"
4363 SCALE_OUT_OUTPUT_XPATH
= "O,/nsr:exec-scale-out"
4365 ACTION
= Enum('ACTION', 'SCALE_IN SCALE_OUT')
4367 def __init__(self
, log
, dts
, loop
, project
, callback
=None):
4368 super().__init
__(log
, dts
, loop
, project
)
4369 self
.callback
= callback
4370 self
.last_instance_id
= defaultdict(int)
4371 self
._regh
_in
= None
4372 self
._regh
_out
= None
4378 self
._log
.warning("RPC already registered for project {}".
4379 format(self
._project
.name
))
4383 def on_scale_in_prepare(xact_info
, action
, ks_path
, msg
):
4384 assert action
== rwdts
.QueryAction
.RPC
4387 if not self
._project
.rpc_check(msg
, xact_info
=xact_info
):
4391 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_IN
)
4393 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleIn
.from_dict({
4394 "instance_id": msg
.instance_id
,
4395 "project_name": self
._project
.name
,})
4397 xact_info
.respond_xpath(
4398 rwdts
.XactRspCode
.ACK
,
4399 self
.__class
__.SCALE_IN_OUTPUT_XPATH
,
4402 except Exception as e
:
4403 self
.log
.exception(e
)
4404 xact_info
.respond_xpath(
4405 rwdts
.XactRspCode
.NACK
,
4406 self
.__class
__.SCALE_IN_OUTPUT_XPATH
)
4409 def on_scale_out_prepare(xact_info
, action
, ks_path
, msg
):
4410 assert action
== rwdts
.QueryAction
.RPC
4413 if not self
._project
.rpc_check(msg
, xact_info
=xact_info
):
4416 scaling_group
= msg
.scaling_group_name_ref
4417 if not msg
.instance_id
:
4418 last_instance_id
= self
.last_instance_id
[scale_group
]
4419 msg
.instance_id
= last_instance_id
+ 1
4420 self
.last_instance_id
[scale_group
] += 1
4423 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_OUT
)
4425 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleOut
.from_dict({
4426 "instance_id": msg
.instance_id
,
4427 "project_name": self
._project
.name
,})
4429 xact_info
.respond_xpath(
4430 rwdts
.XactRspCode
.ACK
,
4431 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
,
4434 except Exception as e
:
4435 self
.log
.exception(e
)
4436 xact_info
.respond_xpath(
4437 rwdts
.XactRspCode
.NACK
,
4438 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
)
4440 scale_in_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4441 on_prepare
=on_scale_in_prepare
)
4442 scale_out_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4443 on_prepare
=on_scale_out_prepare
)
4445 with self
.dts
.group_create() as group
:
4446 self
._regh
_in
= group
.register(
4447 xpath
=self
.__class
__.SCALE_IN_INPUT_XPATH
,
4448 handler
=scale_in_hdl
,
4449 flags
=rwdts
.Flag
.PUBLISHER
)
4450 self
._regh
_out
= group
.register(
4451 xpath
=self
.__class
__.SCALE_OUT_INPUT_XPATH
,
4452 handler
=scale_out_hdl
,
4453 flags
=rwdts
.Flag
.PUBLISHER
)
4455 def deregister(self
):
4456 self
._log
.debug("De-register scale RPCs for project {}".
4457 format(self
._project
.name
))
4459 self
._regh
_in
.deregister()
4460 self
._regh
_in
= None
4462 self
._regh
_out
.deregister()
4463 self
._regh
_out
= None
4466 class NsmProject(ManoProject
):
4468 def __init__(self
, name
, tasklet
, **kw
):
4469 super(NsmProject
, self
).__init
__(tasklet
.log
, name
)
4470 self
.update(tasklet
)
4474 self
._ro
_plugin
_selector
= None
4475 self
._vnffgmgr
= None
4477 self
._nsr
_pub
_handler
= None
4478 self
._vnfr
_pub
_handler
= None
4479 self
._vlr
_pub
_handler
= None
4480 self
._vnfd
_pub
_handler
= None
4481 self
._scale
_cfg
_handler
= None
4483 self
._records
_publisher
_proxy
= None
4487 self
._nsr
_pub
_handler
= publisher
.NsrOpDataDtsHandler(
4488 self
._dts
, self
.log
, self
.loop
, self
)
4489 yield from self
._nsr
_pub
_handler
.register()
4491 self
._vnfr
_pub
_handler
= publisher
.VnfrPublisherDtsHandler(
4492 self
._dts
, self
.log
, self
.loop
, self
)
4493 yield from self
._vnfr
_pub
_handler
.register()
4495 self
._vlr
_pub
_handler
= publisher
.VlrPublisherDtsHandler(
4496 self
._dts
, self
.log
, self
.loop
, self
)
4497 yield from self
._vlr
_pub
_handler
.register()
4499 manifest
= self
._tasklet
.tasklet_info
.get_pb_manifest()
4500 use_ssl
= manifest
.bootstrap_phase
.rwsecurity
.use_ssl
4501 ssl_cert
= manifest
.bootstrap_phase
.rwsecurity
.cert
4502 ssl_key
= manifest
.bootstrap_phase
.rwsecurity
.key
4504 self
._vnfd
_pub
_handler
= publisher
.VnfdPublisher(
4505 use_ssl
, ssl_cert
, ssl_key
, self
.loop
, self
)
4507 self
._records
_publisher
_proxy
= NsmRecordsPublisherProxy(
4512 self
._nsr
_pub
_handler
,
4513 self
._vnfr
_pub
_handler
,
4514 self
._vlr
_pub
_handler
,
4517 # Register the NSM to receive the nsm plugin
4518 # when cloud account is configured
4519 self
._ro
_plugin
_selector
= cloud
.ROAccountPluginSelector(
4524 self
._records
_publisher
_proxy
,
4526 yield from self
._ro
_plugin
_selector
.register()
4528 self
._cloud
_account
_handler
= cloud
.CloudAccountConfigSubscriber(
4535 yield from self
._cloud
_account
_handler
.register()
4537 self
._vnffgmgr
= rwvnffgmgr
.VnffgMgr(self
._dts
, self
.log
, self
.log_hdl
, self
.loop
, self
)
4538 yield from self
._vnffgmgr
.register()
4540 self
._nsm
= NsManager(
4545 self
._nsr
_pub
_handler
,
4546 self
._vnfr
_pub
_handler
,
4547 self
._vlr
_pub
_handler
,
4548 self
._ro
_plugin
_selector
,
4550 self
._vnfd
_pub
_handler
,
4551 self
._cloud
_account
_handler
,
4554 yield from self
._nsm
.register()
4556 def deregister(self
):
4557 self
._log
.debug("Project {} de-register".format(self
.name
))
4558 self
._nsm
.deregister()
4559 self
._vnffgmgr
.deregister()
4560 self
._cloud
_account
_handler
.deregister()
4561 self
._ro
_plugin
_selector
.deregister()
4565 def delete_prepare(self
):
4566 # Check if any NS instance is present
4567 if self
._nsm
and self
._nsm
._nsrs
:
4572 class NsmTasklet(rift
.tasklets
.Tasklet
):
4574 The network service manager tasklet
4576 def __init__(self
, *args
, **kwargs
):
4577 super(NsmTasklet
, self
).__init
__(*args
, **kwargs
)
4578 self
.rwlog
.set_category("rw-mano-log")
4579 self
.rwlog
.set_subcategory("nsm")
4582 self
.project_handler
= None
4590 """ The task start callback """
4591 super(NsmTasklet
, self
).start()
4592 self
.log
.info("Starting NsmTasklet")
4594 self
.log
.debug("Registering with dts")
4595 self
._dts
= rift
.tasklets
.DTS(self
.tasklet_info
,
4596 RwNsmYang
.get_schema(),
4598 self
.on_dts_state_change
)
4600 self
.log
.debug("Created DTS Api GI Object: %s", self
._dts
)
4606 print("Caught Exception in NSM stop:", sys
.exc_info()[0])
4609 def on_instance_started(self
):
4610 """ Task instance started callback """
4611 self
.log
.debug("Got instance started callback")
4615 """ Task init callback """
4616 self
.log
.debug("Got instance started callback")
4618 self
.log
.debug("creating project handler")
4619 self
.project_handler
= ProjectHandler(self
, NsmProject
)
4620 self
.project_handler
.register()
4626 """ Task run callback """
4630 def on_dts_state_change(self
, state
):
4631 """Take action according to current dts state to transition
4632 application into the corresponding application state
4635 state - current dts state
4638 rwdts
.State
.INIT
: rwdts
.State
.REGN_COMPLETE
,
4639 rwdts
.State
.CONFIG
: rwdts
.State
.RUN
,
4643 rwdts
.State
.INIT
: self
.init
,
4644 rwdts
.State
.RUN
: self
.run
,
4647 # Transition application to next state
4648 handler
= handlers
.get(state
, None)
4649 if handler
is not None:
4650 yield from handler()
4652 # Transition dts to next state
4653 next_state
= switch
.get(state
, None)
4654 if next_state
is not None:
4655 self
.log
.debug("Changing state to %s", next_state
)
4656 self
._dts
.handle
.set_state(next_state
)