2 # Copyright 2016 RIFT.IO Inc
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
17 # vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
20 import ncclient
.asyncio_manager
32 from collections
import deque
33 from collections
import defaultdict
37 gi
.require_version('RwYang', '1.0')
38 gi
.require_version('RwNsdYang', '1.0')
39 gi
.require_version('RwDts', '1.0')
40 gi
.require_version('RwNsmYang', '1.0')
41 gi
.require_version('RwNsrYang', '1.0')
42 gi
.require_version('RwTypes', '1.0')
43 gi
.require_version('RwVlrYang', '1.0')
44 gi
.require_version('RwVnfrYang', '1.0')
45 from gi
.repository
import (
61 import rift
.mano
.ncclient
62 import rift
.mano
.config_data
.config
63 import rift
.mano
.dts
as mano_dts
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
= str(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 for conn_p
in self
._vnfd
.connection_point
:
1098 cpr
= VnfrYang
.YangData_RwProject_Project_VnfrCatalog_Vnfr_ConnectionPoint()
1099 cpr
.name
= conn_p
.name
1100 cpr
.type_yang
= conn_p
.type_yang
1101 if conn_p
.has_field('port_security_enabled'):
1102 cpr
.port_security_enabled
= conn_p
.port_security_enabled
1104 vlr_ref
= find_vlr_for_cp(conn_p
)
1106 msg
= "Failed to find VLR for cp = %s" % conn_p
.name
1107 self
._log
.debug("%s", msg
)
1108 # raise VirtualNetworkFunctionRecordError(msg)
1111 cpr
.vlr_ref
= vlr_ref
.id
1112 self
.vnfr_msg
.connection_point
.append(cpr
)
1113 self
._log
.debug("Connection point [%s] added, vnf id=%s vnfd id=%s",
1114 cpr
, self
.vnfr_msg
.id, self
.vnfr_msg
.vnfd
.id)
1116 if not self
.restart_mode
:
1117 yield from self
._dts
.query_create(self
.xpath
,
1121 yield from self
._dts
.query_update(self
.xpath
,
1125 self
._log
.info("Created VNF with xpath %s and vnfr %s",
1126 self
.xpath
, self
.vnfr_msg
)
1128 self
._log
.info("Instantiated VNFR with xpath %s and vnfd %s, vnfr %s",
1129 self
.xpath
, self
._vnfd
, self
.vnfr_msg
)
1132 def update_state(self
, vnfr_msg
):
1133 """ Update this VNFR"""
1134 if vnfr_msg
.operational_status
== "running":
1135 if self
.vnfr_msg
.operational_status
!= "running":
1136 yield from self
.is_active()
1137 elif vnfr_msg
.operational_status
== "failed":
1138 yield from self
.instantiation_failed(failed_reason
=vnfr_msg
.operational_status_details
)
1141 def is_active(self
):
1142 """ This VNFR is active """
1143 self
._log
.debug("VNFR %s is active", self
._vnfr
_id
)
1144 self
.set_state(VnfRecordState
.ACTIVE
)
1147 def instantiation_failed(self
, failed_reason
=None):
1148 """ This VNFR instantiation failed"""
1149 self
._log
.error("VNFR %s instantiation failed", self
._vnfr
_id
)
1150 self
.set_state(VnfRecordState
.FAILED
)
1151 self
._state
_failed
_reason
= failed_reason
1153 def vnfr_in_vnfm(self
):
1154 """ Is there a VNFR record in VNFM """
1155 if (self
._state
== VnfRecordState
.ACTIVE
or
1156 self
._state
== VnfRecordState
.INSTANTIATION_PENDING
or
1157 self
._state
== VnfRecordState
.FAILED
):
1163 def terminate(self
):
1164 """ Terminate this VNF """
1165 if not self
.vnfr_in_vnfm():
1166 self
._log
.debug("Ignoring terminate request for id %s in state %s",
1167 self
.id, self
._state
)
1170 self
._log
.debug("Terminating VNF id:%s", self
.id)
1171 self
.set_state(VnfRecordState
.TERMINATE_PENDING
)
1172 with self
._dts
.transaction(flags
=0) as xact
:
1173 block
= xact
.block_create()
1174 block
.add_query_delete(self
.xpath
)
1175 yield from block
.execute(flags
=0)
1176 self
.set_state(VnfRecordState
.TERMINATED
)
1177 self
._log
.debug("Terminated VNF id:%s", self
.id)
1180 class NetworkServiceStatus(object):
1181 """ A class representing the Network service's status """
1182 MAX_EVENTS_RECORDED
= 10
1183 """ Network service Status class"""
1184 def __init__(self
, dts
, log
, loop
):
1189 self
._state
= NetworkServiceRecordState
.INIT
1190 self
._events
= deque([])
1193 def create_notification(self
, evt
, evt_desc
, evt_details
):
1194 xp
= "N,/rw-nsr:nsm-notification"
1195 notif
= RwNsrYang
.YangNotif_RwNsr_NsmNotification()
1197 notif
.description
= evt_desc
1198 notif
.details
= evt_details
if evt_details
is not None else None
1200 yield from self
._dts
.query_create(xp
, rwdts
.XactFlag
.ADVISE
, notif
)
1201 self
._log
.info("Notification called by creating dts query: %s", notif
)
1203 def record_event(self
, evt
, evt_desc
, evt_details
):
1204 """ Record an event """
1205 self
._log
.debug("Recording event - evt %s, evt_descr %s len = %s",
1206 evt
, evt_desc
, len(self
._events
))
1207 if len(self
._events
) >= NetworkServiceStatus
.MAX_EVENTS_RECORDED
:
1208 self
._events
.popleft()
1209 self
._events
.append((int(time
.time()), evt
, evt_desc
,
1210 evt_details
if evt_details
is not None else None))
1212 self
._loop
.create_task(self
.create_notification(evt
,evt_desc
,evt_details
))
1214 def set_state(self
, state
):
1215 """ set the state of this status object """
1219 """ Return the state as a yang enum string """
1220 state_to_str_map
= {"INIT": "init",
1221 "VL_INIT_PHASE": "vl_init_phase",
1222 "VNF_INIT_PHASE": "vnf_init_phase",
1223 "VNFFG_INIT_PHASE": "vnffg_init_phase",
1224 "SCALING_GROUP_INIT_PHASE": "scaling_group_init_phase",
1225 "RUNNING": "running",
1226 "SCALING_OUT": "scaling_out",
1227 "SCALING_IN": "scaling_in",
1228 "TERMINATE_RCVD": "terminate_rcvd",
1229 "TERMINATE": "terminate",
1230 "VL_TERMINATE_PHASE": "vl_terminate_phase",
1231 "VNF_TERMINATE_PHASE": "vnf_terminate_phase",
1232 "VNFFG_TERMINATE_PHASE": "vnffg_terminate_phase",
1233 "TERMINATED": "terminated",
1235 "VL_INSTANTIATE": "vl_instantiate",
1236 "VL_TERMINATE": "vl_terminate",
1238 return state_to_str_map
[self
._state
.name
]
1242 """ State of this status object """
1247 """ Network Service Record as a message"""
1250 for entry
in self
._events
:
1251 event
= RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_Nsr_OperationalEvents()
1254 event
.timestamp
, event
.event
, event
.description
, event
.details
= entry
1255 event_list
.append(event
)
1259 class NetworkServiceRecord(object):
1260 """ Network service record """
1261 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
1263 def __init__(self
, dts
, log
, loop
, nsm
, nsm_plugin
, nsr_cfg_msg
,
1264 sdn_account_name
, key_pairs
, project
, restart_mode
=False,
1270 self
._nsr
_cfg
_msg
= nsr_cfg_msg
1271 self
._nsm
_plugin
= nsm_plugin
1272 self
._sdn
_account
_name
= sdn_account_name
1273 self
._vlr
_handler
= vlr_handler
1274 self
._project
= project
1277 self
._nsr
_msg
= None
1278 self
._nsr
_regh
= None
1279 self
._key
_pairs
= key_pairs
1284 self
._param
_pools
= {}
1285 self
._scaling
_groups
= {}
1286 self
._create
_time
= int(time
.time())
1287 self
._op
_status
= NetworkServiceStatus(dts
, log
, loop
)
1288 self
._config
_status
= NsrYang
.ConfigStates
.CONFIGURING
1289 self
._config
_status
_details
= None
1291 self
.restart_mode
= restart_mode
1292 self
.config_store
= rift
.mano
.config_data
.config
.ConfigStore(self
._log
)
1293 self
._debug
_running
= False
1294 self
._is
_active
= False
1295 self
._vl
_phase
_completed
= False
1296 self
._vnf
_phase
_completed
= False
1297 self
.vlr_uptime_tasks
= {}
1300 # Initalise the state to init
1301 # The NSR moves through the following transitions
1302 # 1. INIT -> VLS_READY once all the VLs in the NSD are created
1303 # 2. VLS_READY - VNFS_READY when all the VNFs in the NSD are created
1304 # 3. VNFS_READY - READY when the NSR is published
1306 self
.set_state(NetworkServiceRecordState
.INIT
)
1308 self
.substitute_input_parameters
= InputParameterSubstitution(self
._log
, self
._project
)
1311 def nsm_plugin(self
):
1313 return self
._nsm
_plugin
1315 def set_state(self
, state
):
1316 """ Set state for this NSR"""
1317 self
._log
.debug("Setting state to %s", state
)
1318 # We are in init phase and is moving to the next state
1319 # The new state could be a FAILED state or VNF_INIIT_PHASE
1320 if self
.state
== NetworkServiceRecordState
.VL_INIT_PHASE
:
1321 self
._vl
_phase
_completed
= True
1323 if self
.state
== NetworkServiceRecordState
.VNF_INIT_PHASE
:
1324 self
._vnf
_phase
_completed
= True
1326 self
._op
_status
.set_state(state
)
1327 self
._nsm
_plugin
.set_state(self
.id, state
)
1331 """ Get id for this NSR"""
1332 return self
._nsr
_cfg
_msg
.id
1336 """ Name of this network service record """
1337 return self
._nsr
_cfg
_msg
.name
1340 def cloud_account_name(self
):
1341 return self
._nsr
_cfg
_msg
.cloud_account
1344 def om_datacenter_name(self
):
1345 if self
._nsr
_cfg
_msg
.has_field('om_datacenter'):
1346 return self
._nsr
_cfg
_msg
.om_datacenter
1351 """State of this NetworkServiceRecord"""
1352 return self
._op
_status
.state
1356 """ Is this NSR active ?"""
1357 return True if self
._op
_status
.state
== NetworkServiceRecordState
.RUNNING
else False
1361 """ VLRs associated with this NSR"""
1366 """ VNFRs associated with this NSR"""
1371 """ VNFFGRs associated with this NSR"""
1372 return self
._vnffgrs
1375 def scaling_groups(self
):
1376 """ Scaling groups associated with this NSR """
1377 return self
._scaling
_groups
1380 def param_pools(self
):
1381 """ Parameter value pools associated with this NSR"""
1382 return self
._param
_pools
1385 def nsr_cfg_msg(self
):
1386 return self
._nsr
_cfg
_msg
1389 def nsr_cfg_msg(self
, msg
):
1390 self
._nsr
_cfg
_msg
= msg
1394 """ NSD Protobuf for this NSR """
1395 if self
._nsd
is not None:
1397 self
._nsd
= self
._nsr
_cfg
_msg
.nsd
1402 """ NSD ID for this NSR """
1403 return self
.nsd_msg
.id
1407 ''' Get a new job id for config primitive'''
1412 def config_status(self
):
1413 """ Config status for NSR """
1414 return self
._config
_status
1416 def resolve_placement_group_cloud_construct(self
, input_group
):
1418 Returns the cloud specific construct for placement group
1420 copy_dict
= ['name', 'requirement', 'strategy']
1422 for group_info
in self
._nsr
_cfg
_msg
.nsd_placement_group_maps
:
1423 if group_info
.placement_group_ref
== input_group
.name
:
1424 group
= VnfrYang
.YangData_RwProject_Project_VnfrCatalog_Vnfr_PlacementGroupsInfo()
1425 group_dict
= {k
:v
for k
,v
in
1426 group_info
.as_dict().items() if k
!= 'placement_group_ref'}
1427 for param
in copy_dict
:
1428 group_dict
.update({param
: getattr(input_group
, param
)})
1429 group
.from_dict(group_dict
)
1435 return "NSR(name={}, nsd_id={}, cloud_account={})".format(
1436 self
.name
, self
.nsd_id
, self
.cloud_account_name
1439 def _get_vnfd(self
, vnfd_id
, config_xact
):
1440 """ Fetch vnfd msg for the passed vnfd id """
1441 return self
._nsm
.get_vnfd(vnfd_id
, config_xact
)
1443 def _get_vnfd_cloud_account(self
, vnfd_member_index
):
1444 """ Fetch Cloud Account for the passed vnfd id """
1445 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1446 vim_accounts
= [(vnf
.cloud_account
,vnf
.om_datacenter
) for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map \
1447 if vnfd_member_index
== vnf
.member_vnf_index_ref
]
1448 if vim_accounts
and vim_accounts
[0]:
1449 return vim_accounts
[0]
1450 return (self
.cloud_account_name
,self
.om_datacenter_name
)
1452 def _get_constituent_vnfd_msg(self
, vnf_index
):
1453 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
1454 if const_vnfd
.member_vnf_index
== vnf_index
:
1457 raise ValueError("Constituent VNF index %s not found" % vnf_index
)
1459 def record_event(self
, evt
, evt_desc
, evt_details
=None, state
=None):
1460 """ Record an event """
1461 self
._op
_status
.record_event(evt
, evt_desc
, evt_details
)
1462 if state
is not None:
1463 self
.set_state(state
)
1465 def scaling_trigger_str(self
, trigger
):
1466 SCALING_TRIGGER_STRS
= {
1467 NsdYang
.ScalingTrigger
.PRE_SCALE_IN
: 'pre-scale-in',
1468 NsdYang
.ScalingTrigger
.POST_SCALE_IN
: 'post-scale-in',
1469 NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
: 'pre-scale-out',
1470 NsdYang
.ScalingTrigger
.POST_SCALE_OUT
: 'post-scale-out',
1473 return SCALING_TRIGGER_STRS
[trigger
]
1474 except Exception as e
:
1475 self
._log
.error("Scaling trigger mapping error for {} : {}".
1477 self
._log
.exception(e
)
1478 return "Unknown trigger"
1481 def instantiate_vls(self
):
1483 This function instantiates VLs for every VL in this Network Service
1485 self
._log
.debug("Instantiating %d VLs in NSD id %s", len(self
._vlrs
),
1487 for vlr
in self
._vlrs
:
1488 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1489 vlr
.state
= VlRecordState
.ACTIVE
1490 self
.vlr_uptime_tasks
[vlr
.id] = self
._loop
.create_task(self
.vlr_uptime_update(vlr
))
1493 def vlr_uptime_update(self
, vlr
):
1496 vlr_
= RwVlrYang
.YangData_RwProject_Project_VlrCatalog_Vlr
.from_dict({'id': vlr
.id})
1498 vlr_
.uptime
= int(time
.time()) - vlr
._create
_time
1499 xpath
= self
._project
.add_project(VirtualLinkRecord
.vlr_xpath(vlr
))
1500 yield from self
._vlr
_handler
.update(None, xpath
, vlr_
)
1501 yield from asyncio
.sleep(2, loop
=self
._loop
)
1502 except asyncio
.CancelledError
:
1503 self
._log
.debug("Received cancellation request for vlr_uptime_update task")
1504 xpath
= self
._project
.add_project(VirtualLinkRecord
.vlr_xpath(vlr
))
1505 yield from self
._vlr
_handler
.delete(None, xpath
)
1509 def create(self
, config_xact
):
1510 """ Create this network service"""
1511 # Create virtual links for all the external vnf
1512 # connection points in this NS
1513 yield from self
.create_vls()
1515 # Create VNFs in this network service
1516 yield from self
.create_vnfs(config_xact
)
1518 # Create VNFFG for network service
1519 self
.create_vnffgs()
1521 # Create Scaling Groups for each scaling group in NSD
1522 self
.create_scaling_groups()
1524 # Create Parameter Pools
1525 self
.create_param_pools()
1528 def apply_scale_group_config_script(self
, script
, group
, scale_instance
, trigger
, vnfrs
=None):
1529 """ Apply config based on script for scale group """
1532 def add_vnfrs_data(vnfrs_list
):
1533 """ Add as a dict each of the VNFRs data """
1535 for vnfr
in vnfrs_list
:
1536 self
._log
.debug("Add VNFR {} data".format(vnfr
))
1538 vnfr_data
['name'] = vnfr
.name
1539 if trigger
in [NsdYang
.ScalingTrigger
.PRE_SCALE_IN
, NsdYang
.ScalingTrigger
.POST_SCALE_OUT
]:
1540 # Get VNF management and other IPs, etc
1541 opdata
= yield from self
.fetch_vnfr(vnfr
.xpath
)
1542 self
._log
.debug("VNFR {} op data: {}".format(vnfr
.name
, opdata
))
1544 vnfr_data
['rw_mgmt_ip'] = opdata
.mgmt_interface
.ip_address
1545 vnfr_data
['rw_mgmt_port'] = opdata
.mgmt_interface
.port
1546 except Exception as e
:
1547 self
._log
.error("Unable to get management IP for vnfr {}:{}".
1548 format(vnfr
.name
, e
))
1551 vnfr_data
['connection_points'] = []
1552 for cp
in opdata
.connection_point
:
1554 con_pt
['name'] = cp
.name
1555 con_pt
['ip_address'] = cp
.ip_address
1556 vnfr_data
['connection_points'].append(con_pt
)
1557 except Exception as e
:
1558 self
._log
.error("Exception getting connections points for VNFR {}: {}".
1559 format(vnfr
.name
, e
))
1561 vnfrs_data
.append(vnfr_data
)
1562 self
._log
.debug("VNFRs data: {}".format(vnfrs_data
))
1566 def add_nsr_data(nsr
):
1568 nsr_data
['name'] = nsr
.name
1571 if script
is None or len(script
) == 0:
1572 self
._log
.error("Script not provided for scale group config: {}".format(group
.name
))
1575 if script
[0] == '/':
1578 path
= os
.path
.join(os
.environ
['RIFT_INSTALL'], "usr/bin", script
)
1579 if not os
.path
.exists(path
):
1580 self
._log
.error("Config faled for scale group {}: Script does not exist at {}".
1581 format(group
.name
, path
))
1584 # Build a YAML file with all parameters for the script to execute
1585 # The data consists of 5 sections
1587 # 2. Scale group config
1588 # 3. VNFRs in the scale group
1589 # 4. VNFRs outside scale group
1592 data
['trigger'] = group
.trigger_map(trigger
)
1593 data
['config'] = group
.group_msg
.as_dict()
1596 data
["vnfrs_in_group"] = yield from add_vnfrs_data(vnfrs
)
1598 data
["vnfrs_in_group"] = yield from add_vnfrs_data(scale_instance
.vnfrs
)
1600 data
["vnfrs_others"] = yield from add_vnfrs_data(self
.vnfrs
.values())
1601 data
["nsr"] = add_nsr_data(self
)
1604 with tempfile
.NamedTemporaryFile(delete
=False) as tmp_file
:
1605 tmp_file
.write(yaml
.dump(data
, default_flow_style
=True)
1608 self
._log
.debug("Creating a temp file: {} with input data: {}".
1609 format(tmp_file
.name
, data
))
1611 cmd
= "{} {}".format(path
, tmp_file
.name
)
1612 self
._log
.debug("Running the CMD: {}".format(cmd
))
1613 proc
= yield from asyncio
.create_subprocess_shell(cmd
, loop
=self
._loop
)
1614 rc
= yield from proc
.wait()
1616 self
._log
.error("The script {} for scale group {} config returned: {}".
1617 format(script
, group
.name
, rc
))
1625 def apply_scaling_group_config(self
, trigger
, group
, scale_instance
, vnfrs
=None):
1626 """ Apply the config for the scaling group based on trigger """
1627 if group
is None or scale_instance
is None:
1631 def update_config_status(success
=True, err_msg
=None):
1632 self
._log
.debug("Update %s config status to %r : %s",
1633 scale_instance
, success
, err_msg
)
1634 if (scale_instance
.config_status
== "failed"):
1635 # Do not update the config status if it is already in failed state
1638 if scale_instance
.config_status
== "configured":
1639 # Update only to failed state an already configured scale instance
1641 scale_instance
.config_status
= "failed"
1642 scale_instance
.config_err_msg
= err_msg
1643 yield from self
.update_state()
1645 # We are in configuring state
1646 # Only after post scale out mark instance as configured
1647 if trigger
== NsdYang
.ScalingTrigger
.POST_SCALE_OUT
:
1649 scale_instance
.config_status
= "configured"
1651 scale_instance
.config_status
= "failed"
1652 scale_instance
.config_err_msg
= err_msg
1653 yield from self
.update_state()
1655 config
= group
.trigger_config(trigger
)
1659 self
._log
.debug("Scaling group {} config: {}".format(group
.name
, config
))
1660 if config
.has_field("ns_config_primitive_name_ref"):
1661 config_name
= config
.ns_config_primitive_name_ref
1662 nsd_msg
= self
.nsd_msg
1663 config_primitive
= None
1664 for ns_cfg_prim
in nsd_msg
.service_primitive
:
1665 if ns_cfg_prim
.name
== config_name
:
1666 config_primitive
= ns_cfg_prim
1669 if config_primitive
is None:
1670 raise ValueError("Could not find ns_cfg_prim %s in nsr %s" % (config_name
, self
.name
))
1672 self
._log
.debug("Scaling group {} config primitive: {}".format(group
.name
, config_primitive
))
1673 if config_primitive
.has_field("user_defined_script"):
1674 rc
= yield from self
.apply_scale_group_config_script(config_primitive
.user_defined_script
,
1675 group
, scale_instance
, trigger
, vnfrs
)
1678 err_msg
= "Failed config for trigger {} using config script '{}'". \
1679 format(self
.scaling_trigger_str(trigger
),
1680 config_primitive
.user_defined_script
)
1681 yield from update_config_status(success
=rc
, err_msg
=err_msg
)
1684 err_msg
= "Failed config for trigger {} as config script is not specified". \
1685 format(self
.scaling_trigger_str(trigger
))
1686 yield from update_config_status(success
=False, err_msg
=err_msg
)
1687 raise NotImplementedError("Only script based config support for scale group for now: {}".
1690 err_msg
= "Failed config for trigger {} as config primitive is not specified".\
1691 format(self
.scaling_trigger_str(trigger
))
1692 yield from update_config_status(success
=False, err_msg
=err_msg
)
1693 self
._log
.error("Config primitive not specified for config action in scale group %s" %
1697 def create_scaling_groups(self
):
1698 """ This function creates a NSScalingGroup for every scaling
1699 group defined in he NSD"""
1701 for scaling_group_msg
in self
.nsd_msg
.scaling_group_descriptor
:
1702 self
._log
.debug("Found scaling_group %s in nsr id %s",
1703 scaling_group_msg
.name
, self
.id)
1705 group_record
= scale_group
.ScalingGroup(
1710 self
._scaling
_groups
[group_record
.name
] = group_record
1713 def create_scale_group_instance(self
, group_name
, index
, config_xact
, is_default
=False):
1714 group
= self
._scaling
_groups
[group_name
]
1715 scale_instance
= group
.create_instance(index
, is_default
)
1719 self
._log
.debug("Creating %u VNFs associated with NS id %s scaling group %s",
1720 len(self
.nsd_msg
.constituent_vnfd
), self
.id, self
)
1723 for vnf_index
, count
in group
.vnf_index_count_map
.items():
1724 const_vnfd_msg
= self
._get
_constituent
_vnfd
_msg
(vnf_index
)
1725 vnfd_msg
= self
._get
_vnfd
(const_vnfd_msg
.vnfd_id_ref
, config_xact
)
1727 cloud_account_name
, om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd_msg
.member_vnf_index
)
1728 if cloud_account_name
is None:
1729 cloud_account_name
= self
.cloud_account_name
1730 for _
in range(count
):
1731 vnfr
= yield from self
.create_vnf_record(vnfd_msg
, const_vnfd_msg
, cloud_account_name
, om_datacenter_name
, group_name
, index
)
1732 scale_instance
.add_vnfr(vnfr
)
1738 def instantiate_instance():
1739 self
._log
.debug("Creating %s VNFRS", scale_instance
)
1740 vnfrs
= yield from create_vnfs()
1741 yield from self
.publish()
1743 self
._log
.debug("Instantiating %s VNFRS for %s", len(vnfrs
), scale_instance
)
1744 scale_instance
.operational_status
= "vnf_init_phase"
1745 yield from self
.update_state()
1748 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_OUT
,
1749 group
, scale_instance
, vnfrs
)
1751 self
._log
.error("Pre scale out config for scale group {} ({}) failed".
1752 format(group
.name
, index
))
1753 scale_instance
.operational_status
= "failed"
1755 yield from self
.instantiate_vnfs(vnfrs
)
1757 except Exception as e
:
1758 self
._log
.exception("Failed to begin instantiatiation of vnfs for scale group {}: {}".
1759 format(group
.name
, e
))
1760 self
._log
.exception(e
)
1761 scale_instance
.operational_status
= "failed"
1763 yield from self
.update_state()
1765 yield from instantiate_instance()
1768 def delete_scale_group_instance(self
, group_name
, index
):
1769 group
= self
._scaling
_groups
[group_name
]
1770 scale_instance
= group
.get_instance(index
)
1771 if scale_instance
.is_default
:
1772 raise ScalingOperationError("Cannot terminate a default scaling group instance")
1774 scale_instance
.operational_status
= "terminate"
1775 yield from self
.update_state()
1778 def terminate_instance():
1779 self
._log
.debug("Terminating %s VNFRS" % scale_instance
)
1780 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.PRE_SCALE_IN
,
1781 group
, scale_instance
)
1783 self
._log
.error("Pre scale in config for scale group {} ({}) failed".
1784 format(group
.name
, index
))
1786 # Going ahead with terminate, even if there is an error in pre-scale-in config
1787 # as this could be result of scale out failure and we need to cleanup this group
1788 yield from self
.terminate_vnfrs(scale_instance
.vnfrs
)
1789 group
.delete_instance(index
)
1791 scale_instance
.operational_status
= "vnf_terminate_phase"
1792 yield from self
.update_state()
1794 yield from terminate_instance()
1797 def _update_scale_group_instances_status(self
):
1799 def post_scale_out_task(group
, instance
):
1800 # Apply post scale out config once all VNFRs are active
1801 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_OUT
,
1803 instance
.operational_status
= "running"
1805 self
._log
.debug("Scale out for group {} and instance {} succeeded".
1806 format(group
.name
, instance
.instance_id
))
1808 self
._log
.error("Post scale out config for scale group {} ({}) failed".
1809 format(group
.name
, instance
.instance_id
))
1811 yield from self
.update_state()
1813 group_instances
= {group
: group
.instances
for group
in self
._scaling
_groups
.values()}
1814 for group
, instances
in group_instances
.items():
1815 self
._log
.debug("Updating %s instance status", group
)
1816 for instance
in instances
:
1817 instance_vnf_state_list
= [vnfr
.state
for vnfr
in instance
.vnfrs
]
1818 self
._log
.debug("Got vnfr instance states: %s", instance_vnf_state_list
)
1819 if instance
.operational_status
== "vnf_init_phase":
1820 if all([state
== VnfRecordState
.ACTIVE
for state
in instance_vnf_state_list
]):
1821 instance
.operational_status
= "running"
1823 # Create a task for post scale out to allow us to sleep before attempting
1824 # to configure newly created VM's
1825 self
._loop
.create_task(post_scale_out_task(group
, instance
))
1827 elif any([state
== VnfRecordState
.FAILED
for state
in instance_vnf_state_list
]):
1828 self
._log
.debug("Scale out for group {} and instance {} failed".
1829 format(group
.name
, instance
.instance_id
))
1830 instance
.operational_status
= "failed"
1832 elif instance
.operational_status
== "vnf_terminate_phase":
1833 if all([state
== VnfRecordState
.TERMINATED
for state
in instance_vnf_state_list
]):
1834 instance
.operational_status
= "terminated"
1835 rc
= yield from self
.apply_scaling_group_config(NsdYang
.ScalingTrigger
.POST_SCALE_IN
,
1838 self
._log
.debug("Scale in for group {} and instance {} succeeded".
1839 format(group
.name
, instance
.instance_id
))
1841 self
._log
.error("Post scale in config for scale group {} ({}) failed".
1842 format(group
.name
, instance
.instance_id
))
1844 def create_vnffgs(self
):
1845 """ This function creates VNFFGs for every VNFFG in the NSD
1846 associated with this NSR"""
1848 for vnffgd
in self
.nsd_msg
.vnffgd
:
1849 self
._log
.debug("Found vnffgd %s in nsr id %s", vnffgd
, self
.id)
1850 vnffgr
= VnffgRecord(self
._dts
,
1853 self
._nsm
._vnffgmgr
,
1857 self
._sdn
_account
_name
1859 self
._vnffgrs
[vnffgr
.id] = vnffgr
1861 def resolve_vld_ip_profile(self
, nsd_msg
, vld
):
1862 self
._log
.debug("Receieved ip profile ref is %s",vld
.ip_profile_ref
)
1863 if not vld
.has_field('ip_profile_ref'):
1865 profile
= [profile
for profile
in nsd_msg
.ip_profiles
if profile
.name
== vld
.ip_profile_ref
]
1866 return profile
[0] if profile
else None
1869 def _create_vls(self
, vld
, cloud_account
,om_datacenter
):
1870 """Create a VLR in the cloud account specified using the given VLD
1874 cloud_account : Cloud account name
1879 vlr
= yield from VirtualLinkRecord
.create_record(
1888 self
.resolve_vld_ip_profile(self
.nsd_msg
, vld
),
1890 restart_mode
=self
.restart_mode
)
1894 def _extract_cloud_accounts_for_vl(self
, vld
):
1896 Extracts the list of cloud accounts from the NS Config obj
1899 1. Cloud accounts based connection point (vnf_cloud_account_map)
1901 vld : VLD yang object
1906 cloud_account_list
= []
1908 if self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1909 # Handle case where cloud_account is None
1911 for vnf
in self
._nsr
_cfg
_msg
.vnf_cloud_account_map
:
1912 if vnf
.cloud_account
is not None or vnf
.om_datacenter
is not None:
1913 vnf_cloud_map
[vnf
.member_vnf_index_ref
] = (vnf
.cloud_account
,vnf
.om_datacenter
)
1915 for vnfc
in vld
.vnfd_connection_point_ref
:
1916 cloud_account
= vnf_cloud_map
.get(
1917 vnfc
.member_vnf_index_ref
,
1918 (self
.cloud_account_name
,self
.om_datacenter_name
))
1920 cloud_account_list
.append(cloud_account
)
1922 if self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1923 for vld_map
in self
._nsr
_cfg
_msg
.vl_cloud_account_map
:
1924 if vld_map
.vld_id_ref
== vld
.id:
1925 for cloud_account
in vld_map
.cloud_accounts
:
1926 cloud_account_list
.extend((cloud_account
,None))
1927 for om_datacenter
in vld_map
.om_datacenters
:
1928 cloud_account_list
.extend((None,om_datacenter
))
1930 # If no config has been provided then fall-back to the default
1932 if not cloud_account_list
:
1933 cloud_account_list
= [(self
.cloud_account_name
,self
.om_datacenter_name
)]
1935 self
._log
.debug("VL {} cloud accounts: {}".
1936 format(vld
.name
, cloud_account_list
))
1937 return set(cloud_account_list
)
1940 def create_vls(self
):
1941 """ This function creates VLs for every VLD in the NSD
1942 associated with this NSR"""
1943 for vld
in self
.nsd_msg
.vld
:
1945 self
._log
.debug("Found vld %s in nsr id %s", vld
, self
.id)
1946 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1947 for cloud_account
,om_datacenter
in cloud_account_list
:
1948 vlr
= yield from self
._create
_vls
(vld
, cloud_account
,om_datacenter
)
1949 self
._vlrs
.append(vlr
)
1953 def create_vl_instance(self
, vld
):
1954 self
._log
.debug("Create VL for {}: {}".format(self
.id, vld
.as_dict()))
1955 # Check if the VL is already present
1957 for vl
in self
._vlrs
:
1958 if vl
.vld_msg
.id == vld
.id:
1959 self
._log
.debug("The VLD %s already in NSR %s as VLR %s with status %s",
1960 vld
.id, self
.id, vl
.id, vl
.state
)
1962 if vlr
.state
!= VlRecordState
.TERMINATED
:
1963 err_msg
= "VLR for VL %s in NSR %s already instantiated", \
1965 self
._log
.error(err_msg
)
1966 raise NsrVlUpdateError(err_msg
)
1970 cloud_account_list
= self
._extract
_cloud
_accounts
_for
_vl
(vld
)
1971 for account
,om_datacenter
in cloud_account_list
:
1972 vlr
= yield from self
._create
_vls
(vld
, account
,om_datacenter
)
1973 self
._vlrs
.append(vlr
)
1975 vlr
.state
= VlRecordState
.INSTANTIATION_PENDING
1976 yield from self
.update_state()
1979 yield from self
.nsm_plugin
.instantiate_vl(self
, vlr
)
1980 vlr
.state
= VlRecordState
.ACTIVE
1982 except Exception as e
:
1983 err_msg
= "Error instantiating VL for NSR {} and VLD {}: {}". \
1984 format(self
.id, vld
.id, e
)
1985 self
._log
.error(err_msg
)
1986 self
._log
.exception(e
)
1987 vlr
.state
= VlRecordState
.FAILED
1989 yield from self
.update_state()
1992 def delete_vl_instance(self
, vld
):
1993 for vlr
in self
._vlrs
:
1994 if vlr
.vld_msg
.id == vld
.id:
1995 self
._log
.debug("Found VLR %s for VLD %s in NSR %s",
1996 vlr
.id, vld
.id, self
.id)
1997 vlr
.state
= VlRecordState
.TERMINATE_PENDING
1998 yield from self
.update_state()
2001 yield from self
.nsm_plugin
.terminate_vl(vlr
)
2002 vlr
.state
= VlRecordState
.TERMINATED
2003 self
._vlrs
.remove(vlr
)
2005 except Exception as e
:
2006 err_msg
= "Error terminating VL for NSR {} and VLD {}: {}". \
2007 format(self
.id, vld
.id, e
)
2008 self
._log
.error(err_msg
)
2009 self
._log
.exception(e
)
2010 vlr
.state
= VlRecordState
.FAILED
2012 yield from self
.update_state()
2016 def create_vnfs(self
, config_xact
):
2018 This function creates VNFs for every VNF in the NSD
2019 associated with this NSR
2021 self
._log
.debug("Creating %u VNFs associated with this NS id %s",
2022 len(self
.nsd_msg
.constituent_vnfd
), self
.id)
2024 for const_vnfd
in self
.nsd_msg
.constituent_vnfd
:
2025 if not const_vnfd
.start_by_default
:
2026 self
._log
.debug("start_by_default set to False in constituent VNF (%s). Skipping start.",
2027 const_vnfd
.member_vnf_index
)
2030 vnfd_msg
= self
._get
_vnfd
(const_vnfd
.vnfd_id_ref
, config_xact
)
2031 cloud_account_name
,om_datacenter_name
= self
._get
_vnfd
_cloud
_account
(const_vnfd
.member_vnf_index
)
2032 if cloud_account_name
is None:
2033 cloud_account_name
= self
.cloud_account_name
2034 yield from self
.create_vnf_record(vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
)
2037 def get_placement_groups(self
, vnfd_msg
, const_vnfd
):
2038 placement_groups
= []
2039 for group
in self
.nsd_msg
.placement_groups
:
2040 for member_vnfd
in group
.member_vnfd
:
2041 if (member_vnfd
.vnfd_id_ref
== vnfd_msg
.id) and \
2042 (member_vnfd
.member_vnf_index_ref
== const_vnfd
.member_vnf_index
):
2043 group_info
= self
.resolve_placement_group_cloud_construct(group
)
2044 if group_info
is None:
2045 self
._log
.error("Could not resolve cloud-construct for placement group: %s", group
.name
)
2046 ### raise PlacementGroupError("Could not resolve cloud-construct for placement group: {}".format(group.name))
2048 self
._log
.info("Successfully resolved cloud construct for placement group: %s for VNF: %s (Member Index: %s)",
2051 const_vnfd
.member_vnf_index
)
2052 placement_groups
.append(group_info
)
2053 return placement_groups
2056 def create_vnf_record(self
, vnfd_msg
, const_vnfd
, cloud_account_name
, om_datacenter_name
, group_name
=None, group_instance_id
=None):
2057 # Fetch the VNFD associated with this VNF
2058 placement_groups
= self
.get_placement_groups(vnfd_msg
, const_vnfd
)
2059 self
._log
.info("Cloud Account for VNF %d is %s",const_vnfd
.member_vnf_index
,cloud_account_name
)
2060 self
._log
.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s",
2062 const_vnfd
.member_vnf_index
,
2063 [ 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,/nsd:nsd-catalog/nsd:nsd[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,/nsd:nsd-catalog/nsd:nsd[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,/nsd:nsd-catalog/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)",
2827 # Create/Update an NSD record
2828 for cfg
in self
._regh
.get_xact_elements(xact
):
2829 # Only interested in those NSD cfgs whose ID was received in prepare callback
2830 if cfg
.id in scratch
.get('nsds', []) or is_recovery
:
2831 self
._nsm
.update_nsd(cfg
)
2833 scratch
.pop('nsds', None)
2835 return RwTypes
.RwStatus
.SUCCESS
2838 def delete_nsd_libs(nsd_id
):
2839 """ Remove any files uploaded with NSD and stored under $RIFT_ARTIFACTS/libs/<id> """
2841 rift_artifacts_dir
= os
.environ
['RIFT_ARTIFACTS']
2842 nsd_dir
= os
.path
.join(rift_artifacts_dir
, 'launchpad/libs', nsd_id
)
2844 if os
.path
.exists (nsd_dir
):
2845 shutil
.rmtree(nsd_dir
, ignore_errors
=True)
2846 except Exception as e
:
2847 self
._log
.error("Exception in cleaning up NSD libs {}: {}".
2849 self
._log
.exception(e
)
2852 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2853 """ Prepare callback from DTS for NSD config """
2855 self
._log
.info("Got nsd prepare - config received nsd id %s, msg %s",
2858 fref
= ProtobufC
.FieldReference
.alloc()
2859 fref
.goto_whole_message(msg
.to_pbcm())
2861 if fref
.is_field_deleted():
2862 # Delete an NSD record
2863 self
._log
.debug("Deleting NSD with id %s", msg
.id)
2864 if self
._nsm
.nsd_in_use(msg
.id):
2865 self
._log
.debug("Cannot delete NSD in use - %s", msg
.id)
2866 err
= "Cannot delete an NSD in use - %s" % msg
.id
2867 raise NetworkServiceDescriptorRefCountExists(err
)
2869 yield from delete_nsd_libs(msg
.id)
2870 self
._nsm
.delete_nsd(msg
.id)
2872 # Add this NSD to scratch to create/update in apply callback
2873 nsds
= scratch
.setdefault('nsds', [])
2875 # acg._scratch['nsds'].append(msg.id)
2877 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2880 "Registering for NSD config using xpath: %s",
2881 NsdDtsHandler
.XPATH
,
2884 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2885 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2886 # Need a list in scratch to store NSDs to create/update later
2887 # acg._scratch['nsds'] = list()
2888 self
._regh
= acg
.register(
2889 xpath
=self
._project
.add_project(NsdDtsHandler
.XPATH
),
2890 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
2891 on_prepare
=on_prepare
)
2893 def deregister(self
):
2894 self
._log
.debug("De-register NSD handler for project {}".
2895 format(self
._project
.name
))
2897 self
._regh
.deregister()
2901 class VnfdDtsHandler(object):
2902 """ DTS handler for VNFD config changes """
2903 XPATH
= "C,/vnfd:vnfd-catalog/vnfd:vnfd"
2905 def __init__(self
, dts
, log
, loop
, nsm
):
2911 self
._project
= nsm
._project
2915 """ DTS registration handle """
2920 """ Register for VNFD configuration"""
2923 self
._log
.warning("DTS handler already registered for project {}".
2924 format(self
._project
.name
))
2928 def on_apply(dts
, acg
, xact
, action
, scratch
):
2929 """Apply the configuration"""
2930 self
._log
.debug("Got NSM VNFD apply (xact: %s) (action: %s)(scr: %s)",
2931 xact
, action
, scratch
)
2933 # Create/Update a VNFD record
2934 for cfg
in self
._regh
.get_xact_elements(xact
):
2935 # Only interested in those VNFD cfgs whose ID was received in prepare callback
2936 if cfg
.id in scratch
.get('vnfds', []):
2937 self
._nsm
.update_vnfd(cfg
)
2939 for cfg
in self
._regh
.elements
:
2940 if cfg
.id in scratch
.get('deleted_vnfds', []):
2941 yield from self
._nsm
.delete_vnfd(cfg
.id)
2943 scratch
.pop('vnfds', None)
2944 scratch
.pop('deleted_vnfds', None)
2947 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
2948 """ on prepare callback """
2949 self
._log
.debug("Got on prepare for VNFD (path: %s) (action: %s) (msg: %s)",
2950 ks_path
.to_xpath(RwNsmYang
.get_schema()), xact_info
.query_action
, msg
)
2952 fref
= ProtobufC
.FieldReference
.alloc()
2953 fref
.goto_whole_message(msg
.to_pbcm())
2955 # Handle deletes in prepare_callback, but adds/updates in apply_callback
2956 if fref
.is_field_deleted():
2957 self
._log
.debug("Adding msg to deleted field")
2958 deleted_vnfds
= scratch
.setdefault('deleted_vnfds', [])
2959 deleted_vnfds
.append(msg
.id)
2961 # Add this VNFD to scratch to create/update in apply callback
2962 vnfds
= scratch
.setdefault('vnfds', [])
2963 vnfds
.append(msg
.id)
2965 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
2967 xpath
= self
._project
.add_project(VnfdDtsHandler
.XPATH
)
2969 "Registering for VNFD config using xpath {} for project {}"
2970 .format(xpath
, self
._project
))
2971 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
2972 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
2973 # Need a list in scratch to store VNFDs to create/update later
2974 # acg._scratch['vnfds'] = list()
2975 # acg._scratch['deleted_vnfds'] = list()
2976 self
._regh
= acg
.register(
2978 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY
,
2979 on_prepare
=on_prepare
)
2981 def deregister(self
):
2982 self
._log
.debug("De-register VNFD handler for project {}".
2983 format(self
._project
.name
))
2985 self
._regh
.deregister()
2989 class NsrRpcDtsHandler(object):
2990 """ The network service instantiation RPC DTS handler """
2991 EXEC_NSR_CONF_XPATH
= "I,/nsr:start-network-service"
2992 EXEC_NSR_CONF_O_XPATH
= "O,/nsr:start-network-service"
2993 NETCONF_IP_ADDRESS
= "127.0.0.1"
2995 RESTCONF_PORT
= 8888
2996 NETCONF_USER
= "admin"
2997 NETCONF_PW
= "admin"
2998 REST_BASE_V2_URL
= 'https://{}:{}/v2/api/'.format("127.0.0.1",8888)
3000 def __init__(self
, dts
, log
, loop
, nsm
):
3007 self
._ns
_regh
= None
3009 self
._manager
= None
3010 self
._nsr
_config
_url
= NsrRpcDtsHandler
.REST_BASE_V2_URL
+ \
3011 'config/project/{}/ns-instance-config'. \
3012 format(self
._nsm
._project
.name
)
3014 self
._model
= RwYang
.Model
.create_libncx()
3015 self
._model
.load_schema_ypbc(RwNsrYang
.get_schema())
3019 """ Return the NS manager instance """
3023 def wrap_netconf_config_xml(xml
):
3024 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
3028 def _connect(self
, timeout_secs
=240):
3030 start_time
= time
.time()
3031 while (time
.time() - start_time
) < timeout_secs
:
3034 self
._log
.debug("Attemping NsmTasklet netconf connection.")
3036 manager
= yield from ncclient
.asyncio_manager
.asyncio_connect(
3038 host
=NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
,
3039 port
=NsrRpcDtsHandler
.NETCONF_PORT
,
3040 username
=NsrRpcDtsHandler
.NETCONF_USER
,
3041 password
=NsrRpcDtsHandler
.NETCONF_PW
,
3043 look_for_keys
=False,
3044 hostkey_verify
=False,
3049 except ncclient
.transport
.errors
.SSHError
as e
:
3050 self
._log
.warning("Netconf connection to launchpad %s failed: %s",
3051 NsrRpcDtsHandler
.NETCONF_IP_ADDRESS
, str(e
))
3053 yield from asyncio
.sleep(5, loop
=self
._loop
)
3055 raise NsrInstantiationFailed("Failed to connect to Launchpad within %s seconds" %
3058 def _apply_ns_instance_config(self
,payload_dict
):
3059 #self._log.debug("At apply NS instance config with payload %s",payload_dict)
3060 req_hdr
= {'accept':'application/vnd.yang.data+json',
3061 'content-type':'application/vnd.yang.data+json'}
3062 response
=requests
.post(self
._nsr
_config
_url
, headers
=req_hdr
,
3063 auth
=('admin', 'admin'),data
=payload_dict
,verify
=False)
3068 """ Register for NS monitoring read from dts """
3070 self
._log
.warning("RPC already registered for project {}".
3071 format(self
._project
.name
))
3075 def on_ns_config_prepare(xact_info
, action
, ks_path
, msg
):
3076 """ prepare callback from dts start-network-service"""
3077 assert action
== rwdts
.QueryAction
.RPC
3080 if not self
._nsm
._project
.rpc_check(msg
, xact_info
=xact_info
):
3083 rpc_op
= NsrYang
.YangOutput_Nsr_StartNetworkService
.from_dict({
3084 "nsr_id":str(uuid
.uuid4()),
3085 "project_name": msg
.prject_name
,
3088 if not ('name' in rpc_ip
and 'nsd_ref' in rpc_ip
and
3089 ('cloud_account' in rpc_ip
or 'om_datacenter' in rpc_ip
)):
3090 self
._log
.error("Mandatory parameters name or nsd_ref or " +
3091 "cloud account not found in start-network-service {}".
3095 self
._log
.debug("start-network-service RPC input: {}".format(rpc_ip
))
3098 # Add used value to the pool
3099 self
._log
.debug("RPC output: {}".format(rpc_op
))
3101 nsd_copy
= self
.nsm
.get_nsd(rpc_ip
.nsd_ref
)
3103 #if not self._manager:
3104 # self._manager = yield from self._connect()
3106 self
._log
.debug("Configuring ns-instance-config with name %s nsd-ref: %s",
3107 rpc_ip
.name
, rpc_ip
.nsd_ref
)
3109 ns_instance_config_dict
= {"id":rpc_op
.nsr_id
, "admin_status":"ENABLED"}
3110 ns_instance_config_copy_dict
= {k
:v
for k
, v
in rpc_ip
.as_dict().items()
3111 if k
in RwNsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr().fields
}
3112 ns_instance_config_dict
.update(ns_instance_config_copy_dict
)
3114 ns_instance_config
= RwNsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr
.from_dict(ns_instance_config_dict
)
3115 ns_instance_config
.nsd
= NsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr_Nsd()
3116 ns_instance_config
.nsd
.from_dict(nsd_copy
.msg
.as_dict())
3118 payload_dict
= ns_instance_config
.to_json(self
._model
)
3119 #xml = ns_instance_config.to_xml_v2(self._model)
3120 #netconf_xml = self.wrap_netconf_config_xml(xml)
3122 #self._log.debug("Sending configure ns-instance-config xml to %s: %s",
3123 # netconf_xml, NsrRpcDtsHandler.NETCONF_IP_ADDRESS)
3124 self
._log
.debug("Sending configure ns-instance-config json to %s: %s",
3125 self
._nsr
_config
_url
,ns_instance_config
)
3127 #response = yield from self._manager.edit_config(
3129 # config=netconf_xml,
3131 response
= yield from self
._loop
.run_in_executor(
3133 self
._apply
_ns
_instance
_config
,
3136 response
.raise_for_status()
3137 self
._log
.debug("Received edit config response: %s", response
.json())
3139 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
,
3140 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
,
3142 except Exception as e
:
3143 self
._log
.error("Exception processing the "
3144 "start-network-service: {}".format(e
))
3145 self
._log
.exception(e
)
3146 xact_info
.respond_xpath(rwdts
.XactRspCode
.NACK
,
3147 NsrRpcDtsHandler
.EXEC_NSR_CONF_O_XPATH
)
3150 hdl_ns
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_ns_config_prepare
,)
3152 with self
._dts
.group_create() as group
:
3153 self
._ns
_regh
= group
.register(xpath
=NsrRpcDtsHandler
.EXEC_NSR_CONF_XPATH
,
3155 flags
=rwdts
.Flag
.PUBLISHER
,
3158 def deregister(self
):
3159 self
._log
.debug("De-register NSR RPC for project {}".
3160 format(self
._nsm
._project
.name
))
3162 self
._ns
_regh
.deregister()
3163 self
._ns
_regh
= None
3166 class NsrDtsHandler(object):
3167 """ The network service DTS handler """
3168 NSR_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr"
3169 SCALE_INSTANCE_XPATH
= "C,/nsr:ns-instance-config/nsr:nsr/nsr:scaling-group/nsr:instance"
3170 KEY_PAIR_XPATH
= "C,/nsr:key-pair"
3172 def __init__(self
, dts
, log
, loop
, nsm
):
3177 self
._project
= self
._nsm
._project
3179 self
._nsr
_regh
= None
3180 self
._scale
_regh
= None
3181 self
._key
_pair
_regh
= None
3185 """ Return the NS manager instance """
3190 """ Register for Nsr create/update/delete/read requests from dts """
3193 self
._log
.warning("DTS handler already registered for project {}".
3194 format(self
._project
.name
))
3197 def nsr_id_from_keyspec(ks
):
3198 nsr_path_entry
= NsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr
.schema().keyspec_to_entry(ks
)
3199 nsr_id
= nsr_path_entry
.key00
.id
3202 def group_name_from_keyspec(ks
):
3203 group_path_entry
= NsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup
.schema().keyspec_to_entry(ks
)
3204 group_name
= group_path_entry
.key00
.scaling_group_name_ref
3207 def is_instance_in_reg_elements(nsr_id
, group_name
, instance_id
):
3208 """ Return boolean indicating if scaling group instance was already commited previously.
3210 By looking at the existing elements in this registration handle (elements not part
3211 of this current xact), we can tell if the instance was configured previously without
3212 keeping any application state.
3214 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3215 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3216 elem_group_name
= group_name_from_keyspec(keyspec
)
3218 if elem_nsr_id
!= nsr_id
or group_name
!= elem_group_name
:
3221 if instance_cfg
.id == instance_id
:
3226 def get_scale_group_instance_delta(nsr_id
, group_name
, xact
):
3227 delta
= {"added": [], "deleted": []}
3228 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3229 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3230 if elem_nsr_id
!= nsr_id
:
3233 elem_group_name
= group_name_from_keyspec(keyspec
)
3234 if elem_group_name
!= group_name
:
3237 delta
["added"].append(instance_cfg
.id)
3239 for instance_cfg
, keyspec
in self
._scale
_regh
.get_xact_elements(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 if instance_cfg
.id in delta
["added"]:
3249 delta
["added"].remove(instance_cfg
.id)
3251 delta
["deleted"].append(instance_cfg
.id)
3256 def update_nsr_nsd(nsr_id
, xact
, scratch
):
3259 def get_nsr_vl_delta(nsr_id
, xact
, scratch
):
3260 delta
= {"added": [], "deleted": []}
3261 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(xact
, include_keyspec
=True):
3262 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3263 if elem_nsr_id
!= nsr_id
:
3266 if 'vld' in instance_cfg
.nsd
:
3267 for vld
in instance_cfg
.nsd
.vld
:
3268 delta
["added"].append(vld
)
3270 for instance_cfg
, keyspec
in self
._nsr
_regh
.get_xact_elements(include_keyspec
=True):
3271 self
._log
.debug("NSR update: %s", instance_cfg
)
3272 elem_nsr_id
= nsr_id_from_keyspec(keyspec
)
3273 if elem_nsr_id
!= nsr_id
:
3276 if 'vld' in instance_cfg
.nsd
:
3277 for vld
in instance_cfg
.nsd
.vld
:
3278 if vld
in delta
["added"]:
3279 delta
["added"].remove(vld
)
3281 delta
["deleted"].append(vld
)
3285 vl_delta
= yield from get_nsr_vl_delta(nsr_id
, xact
, scratch
)
3286 self
._log
.debug("Got NSR:%s VL instance delta: %s", nsr_id
, vl_delta
)
3288 for vld
in vl_delta
["added"]:
3289 yield from self
._nsm
.nsr_instantiate_vl(nsr_id
, vld
)
3291 for vld
in vl_delta
["deleted"]:
3292 yield from self
._nsm
.nsr_terminate_vl(nsr_id
, vld
)
3294 def get_nsr_key_pairs(dts_member_reg
, xact
):
3296 for instance_cfg
, keyspec
in dts_member_reg
.get_xact_elements(xact
, include_keyspec
=True):
3297 self
._log
.debug("Key pair received is {} KS: {}".format(instance_cfg
, keyspec
))
3298 xpath
= keyspec
.to_xpath(RwNsrYang
.get_schema())
3299 key_pairs
[instance_cfg
.name
] = instance_cfg
3302 def on_apply(dts
, acg
, xact
, action
, scratch
):
3303 """Apply the configuration"""
3304 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3305 xact
, action
, scratch
)
3307 def handle_create_nsr(msg
, key_pairs
=None, restart_mode
=False):
3308 # Handle create nsr requests """
3309 # Do some validations
3310 if not msg
.has_field("nsd"):
3311 err
= "NSD not provided"
3312 self
._log
.error(err
)
3313 raise NetworkServiceRecordError(err
)
3315 self
._log
.debug("Creating NetworkServiceRecord %s from nsr config %s",
3316 msg
.id, msg
.as_dict())
3317 nsr
= self
.nsm
.create_nsr(msg
, key_pairs
=key_pairs
, restart_mode
=restart_mode
)
3320 def handle_delete_nsr(msg
):
3322 def delete_instantiation(ns_id
):
3323 """ Delete instantiation """
3324 with self
._dts
.transaction() as xact
:
3325 yield from self
._nsm
.terminate_ns(ns_id
, xact
)
3327 # Handle delete NSR requests
3328 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3329 # Terminate the NSR instance
3330 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3332 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3333 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3334 nsr
.record_event("terminate-rcvd", event_descr
)
3336 self
._loop
.create_task(delete_instantiation(msg
.id))
3339 def begin_instantiation(nsr
):
3340 # Begin instantiation
3341 self
._log
.info("Beginning NS instantiation: %s", nsr
.id)
3343 yield from self
._nsm
.instantiate_ns(nsr
.id, xact
)
3344 except Exception as e
:
3345 self
._log
.exception("NS instantiation: {}".format(e
))
3348 self
._log
.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
3349 xact
, action
, scratch
)
3351 if action
== rwdts
.AppconfAction
.INSTALL
and xact
.id is None:
3353 for element
in self
._key
_pair
_regh
.elements
:
3354 key_pairs
.append(element
)
3355 for element
in self
._nsr
_regh
.elements
:
3356 nsr
= handle_create_nsr(element
, key_pairs
, restart_mode
=True)
3357 self
._loop
.create_task(begin_instantiation(nsr
))
3360 (added_msgs
, deleted_msgs
, updated_msgs
) = get_add_delete_update_cfgs(self
._nsr
_regh
,
3363 self
._log
.debug("Added: %s, Deleted: %s, Updated: %s", added_msgs
,
3364 deleted_msgs
, updated_msgs
)
3366 for msg
in added_msgs
:
3367 if msg
.id not in self
._nsm
.nsrs
:
3368 self
._log
.info("Create NSR received in on_apply to instantiate NS:%s", msg
.id)
3369 key_pairs
= get_nsr_key_pairs(self
._key
_pair
_regh
, xact
)
3370 nsr
= handle_create_nsr(msg
,key_pairs
)
3371 self
._loop
.create_task(begin_instantiation(nsr
))
3373 for msg
in deleted_msgs
:
3374 self
._log
.info("Delete NSR received in on_apply to terminate NS:%s", msg
.id)
3376 handle_delete_nsr(msg
)
3378 self
._log
.exception("Failed to terminate NS:%s", msg
.id)
3380 for msg
in updated_msgs
:
3381 self
._log
.info("Update NSR received in on_apply: %s", msg
)
3383 self
._nsm
.nsr_update_cfg(msg
.id, msg
)
3386 self
._loop
.create_task(update_nsr_nsd(msg
.id, xact
, scratch
))
3388 for group
in msg
.scaling_group
:
3389 instance_delta
= get_scale_group_instance_delta(msg
.id, group
.scaling_group_name_ref
, xact
)
3390 self
._log
.debug("Got NSR:%s scale group instance delta: %s", msg
.id, instance_delta
)
3392 for instance_id
in instance_delta
["added"]:
3393 self
._nsm
.scale_nsr_out(msg
.id, group
.scaling_group_name_ref
, instance_id
, xact
)
3395 for instance_id
in instance_delta
["deleted"]:
3396 self
._nsm
.scale_nsr_in(msg
.id, group
.scaling_group_name_ref
, instance_id
)
3399 return RwTypes
.RwStatus
.SUCCESS
3402 def on_prepare(dts
, acg
, xact
, xact_info
, ks_path
, msg
, scratch
):
3403 """ Prepare calllback from DTS for NSR """
3405 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3406 action
= xact_info
.query_action
3408 "Got Nsr prepare callback (xact: %s) (action: %s) (info: %s), %s:%s)",
3409 xact
, action
, xact_info
, xpath
, msg
3413 def delete_instantiation(ns_id
):
3414 """ Delete instantiation """
3415 yield from self
._nsm
.terminate_ns(ns_id
, None)
3417 def handle_delete_nsr():
3418 """ Handle delete NSR requests """
3419 self
._log
.info("Delete req for NSR Id: %s received", msg
.id)
3420 # Terminate the NSR instance
3421 nsr
= self
._nsm
.get_ns_by_nsr_id(msg
.id)
3423 nsr
.set_state(NetworkServiceRecordState
.TERMINATE_RCVD
)
3424 event_descr
= "Terminate rcvd for NS Id:%s" % msg
.id
3425 nsr
.record_event("terminate-rcvd", event_descr
)
3427 self
._loop
.create_task(delete_instantiation(msg
.id))
3429 fref
= ProtobufC
.FieldReference
.alloc()
3430 fref
.goto_whole_message(msg
.to_pbcm())
3432 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
, rwdts
.QueryAction
.DELETE
]:
3433 # if this is an NSR create
3434 if action
!= rwdts
.QueryAction
.DELETE
and msg
.id not in self
._nsm
.nsrs
:
3435 # Ensure the Cloud account/datacenter has been specified
3436 if not msg
.has_field("cloud_account") and not msg
.has_field("om_datacenter"):
3437 raise NsrInstantiationFailed("Cloud account or datacenter not specified in NSR")
3439 # Check if nsd is specified
3440 if not msg
.has_field("nsd"):
3441 raise NsrInstantiationFailed("NSD not specified in NSR")
3444 nsr
= self
._nsm
.nsrs
[msg
.id]
3446 if msg
.has_field("nsd"):
3447 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3448 raise NsrVlUpdateError("Unable to update VL when NSR not in running state")
3449 if 'vld' not in msg
.nsd
or len(msg
.nsd
.vld
) == 0:
3450 raise NsrVlUpdateError("NS config NSD should have atleast 1 VLD defined")
3452 if msg
.has_field("scaling_group"):
3453 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3454 raise ScalingOperationError("Unable to perform scaling action when NS is not in running state")
3456 if len(msg
.scaling_group
) > 1:
3457 raise ScalingOperationError("Only a single scaling group can be configured at a time")
3459 for group_msg
in msg
.scaling_group
:
3460 num_new_group_instances
= len(group_msg
.instance
)
3461 if num_new_group_instances
> 1:
3462 raise ScalingOperationError("Only a single scaling instance can be modified at a time")
3464 elif num_new_group_instances
== 1:
3465 scale_group
= nsr
.scaling_groups
[group_msg
.scaling_group_name_ref
]
3466 if action
in [rwdts
.QueryAction
.CREATE
, rwdts
.QueryAction
.UPDATE
]:
3467 if len(scale_group
.instances
) == scale_group
.max_instance_count
:
3468 raise ScalingOperationError("Max instances for %s reached" % scale_group
)
3470 acg
.handle
.prepare_complete_ok(xact_info
.handle
)
3473 xpath
= self
._project
.add_project(NsrDtsHandler
.NSR_XPATH
)
3474 self
._log
.debug("Registering for NSR config using xpath: {}".
3477 acg_hdl
= rift
.tasklets
.AppConfGroup
.Handler(on_apply
=on_apply
)
3478 with self
._dts
.appconf_group_create(handler
=acg_hdl
) as acg
:
3479 self
._nsr
_regh
= acg
.register(
3481 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3482 on_prepare
=on_prepare
3485 self
._scale
_regh
= acg
.register(
3486 xpath
=self
._project
.add_project(NsrDtsHandler
.SCALE_INSTANCE_XPATH
),
3487 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY| rwdts
.Flag
.CACHE
,
3490 self
._key
_pair
_regh
= acg
.register(
3491 xpath
=self
._project
.add_project(NsrDtsHandler
.KEY_PAIR_XPATH
),
3492 flags
=rwdts
.Flag
.SUBSCRIBER | rwdts
.Flag
.DELTA_READY | rwdts
.Flag
.CACHE
,
3495 def deregister(self
):
3496 self
._log
.debug("De-register NSR config for project {}".
3497 format(self
._project
.name
))
3499 self
._nsr
_regh
.deregister()
3500 self
._nsr
_regh
= None
3501 if self
._scale
_regh
:
3502 self
._scale
_regh
.deregister()
3503 self
._scale
_regh
= None
3504 if self
._key
_pair
_regh
:
3505 self
._key
_pair
_regh
.deregister()
3506 self
._key
_pair
_regh
= None
3509 class NsrOpDataDtsHandler(object):
3510 """ The network service op data DTS handler """
3511 XPATH
= "D,/nsr:ns-instance-opdata/nsr:nsr"
3513 def __init__(self
, dts
, log
, loop
, nsm
):
3519 self
._project
= nsm
._project
3524 """ Return the registration handle"""
3529 """ Return the NS manager instance """
3534 """ Register for Nsr op data publisher registration"""
3536 self
._log
.warning("NSR op data handler already registered for project {}".
3537 format(self
._project
.name
))
3540 xpath
= self
._project
.add_project(NsrOpDataDtsHandler
.XPATH
)
3541 self
._log
.debug("Registering Nsr op data path {} as publisher".
3544 hdl
= rift
.tasklets
.DTS
.RegistrationHandler()
3545 handlers
= rift
.tasklets
.Group
.Handler()
3546 with self
._dts
.group_create(handler
=handlers
) as group
:
3547 self
._regh
= group
.register(xpath
=xpath
,
3549 flags
=rwdts
.Flag
.PUBLISHER | rwdts
.Flag
.NO_PREP_READ | rwdts
.Flag
.DATASTORE
)
3551 def deregister(self
):
3552 self
._log
.debug("De-register NSR opdata for project {}".
3553 format(self
._project
.name
))
3555 self
._regh
.deregister()
3559 def create(self
, xpath
, msg
):
3561 Create an NS record in DTS with the path and message
3563 path
= self
._project
.add_project(xpath
)
3564 self
._log
.debug("Creating NSR %s:%s", path
, msg
)
3565 self
.regh
.create_element(path
, msg
)
3566 self
._log
.debug("Created NSR, %s:%s", path
, msg
)
3569 def update(self
, xpath
, msg
, flags
=rwdts
.XactFlag
.REPLACE
):
3571 Update an NS record in DTS with the path and message
3573 path
= self
._project
.add_project(xpath
)
3574 self
._log
.debug("Updating NSR, %s:%s regh = %s", path
, msg
, self
.regh
)
3575 self
.regh
.update_element(path
, msg
, flags
)
3576 self
._log
.debug("Updated NSR, %s:%s", path
, msg
)
3579 def delete(self
, xpath
):
3581 Update an NS record in DTS with the path and message
3583 path
= self
._project
.add_project(xpath
)
3584 self
._log
.debug("Deleting NSR path:%s", path
)
3585 self
.regh
.delete_element(path
)
3586 self
._log
.debug("Deleted NSR path:%s", path
)
3589 class VnfrDtsHandler(object):
3590 """ The virtual network service DTS handler """
3591 XPATH
= "D,/vnfr:vnfr-catalog/vnfr:vnfr"
3593 def __init__(self
, dts
, log
, loop
, nsm
):
3603 """ Return registration handle """
3608 """ Return the NS manager instance """
3613 """ Register for vnfr create/update/delete/ advises from dts """
3615 self
._log
.warning("VNFR DTS handler already registered for project {}".
3616 format(self
._project
.name
))
3620 def on_commit(xact_info
):
3621 """ The transaction has been committed """
3622 self
._log
.debug("Got vnfr commit (xact_info: %s)", xact_info
)
3623 return rwdts
.MemberRspCode
.ACTION_OK
3626 def on_prepare(xact_info
, action
, ks_path
, msg
):
3627 """ prepare callback from dts """
3628 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3630 "Got vnfr on_prepare cb (xact_info: %s, action: %s): %s:%s",
3631 xact_info
, action
, ks_path
, msg
3634 schema
= VnfrYang
.YangData_RwProject_Project_VnfrCatalog_Vnfr
.schema()
3635 path_entry
= schema
.keyspec_to_entry(ks_path
)
3636 if path_entry
.key00
.id not in self
._nsm
._vnfrs
:
3637 # Check if this is a monitoring param xpath
3638 if 'vnfr:monitoring-param' not in xpath
:
3639 self
._log
.error("%s request for non existent record path %s",
3641 xact_info
.respond_xpath(rwdts
.XactRspCode
.NA
, xpath
)
3645 if action
== rwdts
.QueryAction
.CREATE
or action
== rwdts
.QueryAction
.UPDATE
:
3646 yield from self
._nsm
.update_vnfr(msg
)
3647 elif action
== rwdts
.QueryAction
.DELETE
:
3648 self
._log
.debug("Deleting VNFR with id %s", path_entry
.key00
.id)
3649 self
._nsm
.delete_vnfr(path_entry
.key00
.id)
3651 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
, xpath
)
3653 self
._log
.debug("Registering for VNFR using xpath: %s",
3654 VnfrDtsHandler
.XPATH
,)
3656 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_commit
=on_commit
,
3657 on_prepare
=on_prepare
,)
3658 with self
._dts
.group_create() as group
:
3659 self
._regh
= group
.register(xpath
=self
._nsm
._project
.add_project(
3660 VnfrDtsHandler
.XPATH
),
3662 flags
=(rwdts
.Flag
.SUBSCRIBER
),)
3664 def deregister(self
):
3665 self
._log
.debug("De-register VNFR for project {}".
3666 format(self
._project
.name
))
3668 self
._regh
.deregister()
3671 class NsdRefCountDtsHandler(object):
3672 """ The NSD Ref Count DTS handler """
3673 XPATH
= "D,/nsr:ns-instance-opdata/rw-nsr:nsd-ref-count"
3675 def __init__(self
, dts
, log
, loop
, nsm
):
3685 """ Return registration handle """
3690 """ Return the NS manager instance """
3695 """ Register for NSD ref count read from dts """
3697 self
._log
.warning("NSD ref DTS handler already registered for project {}".
3698 format(self
._project
.name
))
3703 def on_prepare(xact_info
, action
, ks_path
, msg
):
3704 """ prepare callback from dts """
3705 xpath
= ks_path
.to_xpath(RwNsrYang
.get_schema())
3707 if action
== rwdts
.QueryAction
.READ
:
3708 schema
= RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount
.schema()
3709 path_entry
= schema
.keyspec_to_entry(ks_path
)
3710 nsd_list
= yield from self
._nsm
.get_nsd_refcount(path_entry
.key00
.nsd_id_ref
)
3711 for xpath
, msg
in nsd_list
:
3712 xact_info
.respond_xpath(rsp_code
=rwdts
.XactRspCode
.MORE
,
3715 xact_info
.respond_xpath(rwdts
.XactRspCode
.ACK
)
3717 raise NetworkServiceRecordError("Not supported operation %s" % action
)
3719 hdl
= rift
.tasklets
.DTS
.RegistrationHandler(on_prepare
=on_prepare
,)
3720 with self
._dts
.group_create() as group
:
3721 self
._regh
= group
.register(xpath
=self
._nsm
._project
.add_project(
3722 NsdRefCountDtsHandler
.XPATH
),
3724 flags
=rwdts
.Flag
.PUBLISHER
,)
3726 def deregister(self
):
3727 self
._log
.debug("De-register NSD Ref count for project {}".
3728 format(self
._project
.name
))
3730 self
._regh
.deregister()
3734 class NsManager(object):
3735 """ The Network Service Manager class"""
3736 def __init__(self
, dts
, log
, loop
, project
,
3737 nsr_handler
, vnfr_handler
, vlr_handler
, ro_plugin_selector
,
3738 vnffgmgr
, vnfd_pub_handler
, cloud_account_handler
):
3742 self
._project
= project
3743 self
._nsr
_handler
= nsr_handler
3744 self
._vnfr
_pub
_handler
= vnfr_handler
3745 self
._vlr
_pub
_handler
= vlr_handler
3746 self
._vnffgmgr
= vnffgmgr
3747 self
._vnfd
_pub
_handler
= vnfd_pub_handler
3748 self
._cloud
_account
_handler
= cloud_account_handler
3750 self
._ro
_plugin
_selector
= ro_plugin_selector
3751 self
._ncclient
= rift
.mano
.ncclient
.NcClient(
3763 self
.cfgmgr_obj
= conman
.ROConfigManager(log
, loop
, dts
, self
)
3765 # TODO: All these handlers should move to tasklet level.
3766 # Passing self is often an indication of bad design
3767 self
._nsd
_dts
_handler
= NsdDtsHandler(dts
, log
, loop
, self
)
3768 self
._vnfd
_dts
_handler
= VnfdDtsHandler(dts
, log
, loop
, self
)
3769 self
._dts
_handlers
= [self
._nsd
_dts
_handler
,
3770 VnfrDtsHandler(dts
, log
, loop
, self
),
3771 NsdRefCountDtsHandler(dts
, log
, loop
, self
),
3772 NsrDtsHandler(dts
, log
, loop
, self
),
3773 ScalingRpcHandler(log
, dts
, loop
, self
._project
,
3774 self
.scale_rpc_callback
),
3775 NsrRpcDtsHandler(dts
, log
, loop
, self
),
3776 self
._vnfd
_dts
_handler
,
3797 def nsr_handler(self
):
3798 """" NSR handler """
3799 return self
._nsr
_handler
3803 """" So Obj handler """
3808 """ NSRs in this NSM"""
3813 """ NSDs in this NSM"""
3818 """ VNFDs in this NSM"""
3823 """ VNFRs in this NSM"""
3827 def nsr_pub_handler(self
):
3828 """ NSR publication handler """
3829 return self
._nsr
_handler
3832 def vnfr_pub_handler(self
):
3833 """ VNFR publication handler """
3834 return self
._vnfr
_pub
_handler
3837 def vlr_pub_handler(self
):
3838 """ VLR publication handler """
3839 return self
._vlr
_pub
_handler
3842 def vnfd_pub_handler(self
):
3843 return self
._vnfd
_pub
_handler
3847 """ Register all static DTS handlers """
3848 for dts_handle
in self
._dts
_handlers
:
3849 yield from dts_handle
.register()
3851 def deregister(self
):
3852 """ Register all static DTS handlers """
3853 for dts_handle
in self
._dts
_handlers
:
3854 yield from dts_handle
.deregister()
3857 def get_ns_by_nsr_id(self
, nsr_id
):
3858 """ get NSR by nsr id """
3859 if nsr_id
not in self
._nsrs
:
3860 raise NetworkServiceRecordError("NSR id %s not found" % nsr_id
)
3862 return self
._nsrs
[nsr_id
]
3864 def scale_nsr_out(self
, nsr_id
, scale_group_name
, instance_id
, config_xact
):
3865 self
.log
.debug("Scale out NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3870 nsr
= self
._nsrs
[nsr_id
]
3871 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3872 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3874 self
._loop
.create_task(nsr
.create_scale_group_instance(scale_group_name
, instance_id
, config_xact
))
3876 def scale_nsr_in(self
, nsr_id
, scale_group_name
, instance_id
):
3877 self
.log
.debug("Scale in NetworkServiceRecord (nsr_id: %s) (scaling group: %s) (instance_id: %s)",
3882 nsr
= self
._nsrs
[nsr_id
]
3883 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3884 raise ScalingOperationError("Cannot perform scaling operation if NSR is not in running state")
3886 self
._loop
.create_task(nsr
.delete_scale_group_instance(scale_group_name
, instance_id
))
3888 def scale_rpc_callback(self
, xact
, msg
, action
):
3889 """Callback handler for RPC calls
3891 xact : Transaction Handler
3893 action : Scaling Action
3895 ScalingGroupInstance
= NsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup_Instance
3896 ScalingGroup
= NsrYang
.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup
3898 xpath
= self
._project
.add_project(
3899 ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').
3900 format(msg
.nsr_id_ref
))
3902 instance
= ScalingGroupInstance
.from_dict({
3903 "id": msg
.instance_id
,
3904 "project_name": self
._project
.name
,})
3907 def get_nsr_scaling_group():
3908 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
3910 for result
in results
:
3911 res
= yield from result
3912 nsr_config
= res
.result
3914 for scaling_group
in nsr_config
.scaling_group
:
3915 if scaling_group
.scaling_group_name_ref
== msg
.scaling_group_name_ref
:
3918 scaling_group
= nsr_config
.scaling_group
.add()
3919 scaling_group
.scaling_group_name_ref
= msg
.scaling_group_name_ref
3921 return (nsr_config
, scaling_group
)
3924 def update_config(nsr_config
):
3925 xml
= self
._ncclient
.convert_to_xml(RwNsrYang
, nsr_config
)
3926 xml
= '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml
)
3927 yield from self
._ncclient
.connect()
3928 yield from self
._ncclient
.manager
.edit_config(target
="running", config
=xml
, default_operation
="replace")
3932 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3933 scaling_group
.instance
.append(instance
)
3934 yield from update_config(nsr_config
)
3938 nsr_config
, scaling_group
= yield from get_nsr_scaling_group()
3939 scaling_group
.instance
.remove(instance
)
3940 yield from update_config(nsr_config
)
3942 if action
== ScalingRpcHandler
.ACTION
.SCALE_OUT
:
3943 self
._loop
.create_task(scale_out())
3945 self
._loop
.create_task(scale_in())
3947 # Opdata based calls, disabled for now!
3948 # if action == ScalingRpcHandler.ACTION.SCALE_OUT:
3949 # self.scale_nsr_out(
3951 # msg.scaling_group_name_ref,
3955 # self.scale_nsr_in(
3957 # msg.scaling_group_name_ref,
3960 def nsr_update_cfg(self
, nsr_id
, msg
):
3961 nsr
= self
._nsrs
[nsr_id
]
3962 nsr
.nsr_cfg_msg
= msg
3964 def nsr_instantiate_vl(self
, nsr_id
, vld
):
3965 self
.log
.debug("NSR {} create VL {}".format(nsr_id
, vld
))
3966 nsr
= self
._nsrs
[nsr_id
]
3967 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3968 raise NsrVlUpdateError("Cannot perform VL instantiate if NSR is not in running state")
3970 # Not calling in a separate task as this is called from a separate task
3971 yield from nsr
.create_vl_instance(vld
)
3973 def nsr_terminate_vl(self
, nsr_id
, vld
):
3974 self
.log
.debug("NSR {} delete VL {}".format(nsr_id
, vld
.id))
3975 nsr
= self
._nsrs
[nsr_id
]
3976 if nsr
.state
!= NetworkServiceRecordState
.RUNNING
:
3977 raise NsrVlUpdateError("Cannot perform VL terminate if NSR is not in running state")
3979 # Not calling in a separate task as this is called from a separate task
3980 yield from nsr
.delete_vl_instance(vld
)
3982 def create_nsr(self
, nsr_msg
, key_pairs
=None,restart_mode
=False):
3983 """ Create an NSR instance """
3984 if nsr_msg
.id in self
._nsrs
:
3985 msg
= "NSR id %s already exists" % nsr_msg
.id
3986 self
._log
.error(msg
)
3987 raise NetworkServiceRecordError(msg
)
3989 self
._log
.info("Create NetworkServiceRecord nsr id %s from nsd_id %s",
3993 nsm_plugin
= self
._ro
_plugin
_selector
.ro_plugin
3994 sdn_account_name
= self
._cloud
_account
_handler
.get_cloud_account_sdn_name(nsr_msg
.cloud_account
)
3996 nsr
= NetworkServiceRecord(self
._dts
,
4005 restart_mode
=restart_mode
,
4006 vlr_handler
=self
._ro
_plugin
_selector
._records
_publisher
._vlr
_pub
_hdlr
4008 self
._nsrs
[nsr_msg
.id] = nsr
4009 nsm_plugin
.create_nsr(nsr_msg
, nsr_msg
.nsd
, key_pairs
)
4013 def delete_nsr(self
, nsr_id
):
4015 Delete NSR with the passed nsr id
4017 del self
._nsrs
[nsr_id
]
4020 def instantiate_ns(self
, nsr_id
, config_xact
):
4021 """ Instantiate an NS instance """
4022 self
._log
.debug("Instantiating Network service id %s", nsr_id
)
4023 if nsr_id
not in self
._nsrs
:
4024 err
= "NSR id %s not found " % nsr_id
4025 self
._log
.error(err
)
4026 raise NetworkServiceRecordError(err
)
4028 nsr
= self
._nsrs
[nsr_id
]
4029 yield from nsr
.nsm_plugin
.instantiate_ns(nsr
, config_xact
)
4032 def update_vnfr(self
, vnfr
):
4033 """Create/Update an VNFR """
4035 vnfr_state
= self
._vnfrs
[vnfr
.id].state
4036 self
._log
.debug("Updating VNFR with state %s: vnfr %s", vnfr_state
, vnfr
)
4038 yield from self
._vnfrs
[vnfr
.id].update_state(vnfr
)
4039 nsr
= self
.find_nsr_for_vnfr(vnfr
.id)
4040 yield from nsr
.update_state()
4042 def find_nsr_for_vnfr(self
, vnfr_id
):
4043 """ Find the NSR which )has the passed vnfr id"""
4044 for nsr
in list(self
.nsrs
.values()):
4045 for vnfr
in list(nsr
.vnfrs
.values()):
4046 if vnfr
.id == vnfr_id
:
4050 def delete_vnfr(self
, vnfr_id
):
4051 """ Delete VNFR with the passed id"""
4052 del self
._vnfrs
[vnfr_id
]
4054 def get_nsd_ref(self
, nsd_id
):
4055 """ Get network service descriptor for the passed nsd_id
4057 nsd
= self
.get_nsd(nsd_id
)
4062 def get_nsr_config(self
, nsd_id
):
4063 xpath
= self
._project
.add_project("C,/nsr:ns-instance-config")
4064 results
= yield from self
._dts
.query_read(xpath
, rwdts
.XactFlag
.MERGE
)
4066 for result
in results
:
4067 entry
= yield from result
4068 ns_instance_config
= entry
.result
4070 for nsr
in ns_instance_config
.nsr
:
4071 if nsr
.nsd
.id == nsd_id
:
4077 def nsd_unref_by_nsr_id(self
, nsr_id
):
4078 """ Unref the network service descriptor based on NSR id """
4079 self
._log
.debug("NSR Unref called for Nsr Id:%s", nsr_id
)
4080 if nsr_id
in self
._nsrs
:
4081 nsr
= self
._nsrs
[nsr_id
]
4084 nsd
= self
.get_nsd(nsr
.nsd_id
)
4085 self
._log
.debug("Releasing ref on NSD %s held by NSR %s - Curr %d",
4086 nsd
.id, nsr
.id, nsd
.ref_count
)
4088 except NetworkServiceDescriptorError
:
4089 # We store a copy of NSD in NSR and the NSD in nsd-catalog
4094 self
._log
.error("Cannot find NSR with id %s", nsr_id
)
4095 raise NetworkServiceDescriptorUnrefError("No NSR with id" % nsr_id
)
4098 def nsd_unref(self
, nsd_id
):
4099 """ Unref the network service descriptor associated with the id """
4100 nsd
= self
.get_nsd(nsd_id
)
4103 def get_nsd(self
, nsd_id
):
4104 """ Get network service descriptor for the passed nsd_id"""
4105 if nsd_id
not in self
._nsds
:
4106 self
._log
.error("Cannot find NSD id:%s", nsd_id
)
4107 raise NetworkServiceDescriptorError("Cannot find NSD id:%s", nsd_id
)
4109 return self
._nsds
[nsd_id
]
4111 def create_nsd(self
, nsd_msg
):
4112 """ Create a network service descriptor """
4113 self
._log
.debug("Create network service descriptor - %s", nsd_msg
)
4114 if nsd_msg
.id in self
._nsds
:
4115 self
._log
.error("Cannot create NSD %s -NSD ID already exists", nsd_msg
)
4116 raise NetworkServiceDescriptorError("NSD already exists-%s", nsd_msg
.id)
4118 nsd
= NetworkServiceDescriptor(
4125 self
._nsds
[nsd_msg
.id] = nsd
4129 def update_nsd(self
, nsd
):
4130 """ update the Network service descriptor """
4131 self
._log
.debug("Update network service descriptor - %s", nsd
)
4132 if nsd
.id not in self
._nsds
:
4133 self
._log
.debug("No NSD found - creating NSD id = %s", nsd
.id)
4134 self
.create_nsd(nsd
)
4136 self
._log
.debug("Updating NSD id = %s, nsd = %s", nsd
.id, nsd
)
4137 self
._nsds
[nsd
.id].update(nsd
)
4139 def delete_nsd(self
, nsd_id
):
4140 """ Delete the Network service descriptor with the passed id """
4141 self
._log
.debug("Deleting the network service descriptor - %s", nsd_id
)
4142 if nsd_id
not in self
._nsds
:
4143 self
._log
.debug("Delete NSD failed - cannot find nsd-id %s", nsd_id
)
4144 raise NetworkServiceDescriptorNotFound("Cannot find %s", nsd_id
)
4146 if nsd_id
not in self
._nsds
:
4147 self
._log
.debug("Cannot delete NSD id %s reference exists %s",
4149 self
._nsds
[nsd_id
].ref_count
)
4150 raise NetworkServiceDescriptorRefCountExists(
4151 "Cannot delete :%s, ref_count:%s",
4153 self
._nsds
[nsd_id
].ref_count
)
4155 del self
._nsds
[nsd_id
]
4157 def get_vnfd_config(self
, xact
):
4158 vnfd_dts_reg
= self
._vnfd
_dts
_handler
.regh
4159 for cfg
in vnfd_dts_reg
.get_xact_elements(xact
):
4160 if cfg
.id not in self
._vnfds
:
4161 self
.create_vnfd(cfg
)
4163 def get_vnfd(self
, vnfd_id
, xact
):
4164 """ Get virtual network function descriptor for the passed vnfd_id"""
4165 if vnfd_id
not in self
._vnfds
:
4166 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4167 self
.get_vnfd_config(xact
)
4169 if vnfd_id
not in self
._vnfds
:
4170 self
._log
.error("Cannot find VNFD id:%s", vnfd_id
)
4171 raise VnfDescriptorError("Cannot find VNFD id:%s", vnfd_id
)
4173 return self
._vnfds
[vnfd_id
]
4175 def create_vnfd(self
, vnfd
):
4176 """ Create a virtual network function descriptor """
4177 self
._log
.debug("Create virtual network function descriptor - %s", vnfd
)
4178 if vnfd
.id in self
._vnfds
:
4179 self
._log
.error("Cannot create VNFD %s -VNFD ID already exists", vnfd
)
4180 raise VnfDescriptorError("VNFD already exists-%s", vnfd
.id)
4182 self
._vnfds
[vnfd
.id] = vnfd
4183 return self
._vnfds
[vnfd
.id]
4185 def update_vnfd(self
, vnfd
):
4186 """ Update the virtual network function descriptor """
4187 self
._log
.debug("Update virtual network function descriptor- %s", vnfd
)
4190 if vnfd
.id not in self
._vnfds
:
4191 self
._log
.debug("No VNFD found - creating VNFD id = %s", vnfd
.id)
4192 self
.create_vnfd(vnfd
)
4194 self
._log
.debug("Updating VNFD id = %s, vnfd = %s", vnfd
.id, vnfd
)
4195 self
._vnfds
[vnfd
.id] = vnfd
4198 def delete_vnfd(self
, vnfd_id
):
4199 """ Delete the virtual network function descriptor with the passed id """
4200 self
._log
.debug("Deleting the virtual network function descriptor - %s", vnfd_id
)
4201 if vnfd_id
not in self
._vnfds
:
4202 self
._log
.debug("Delete VNFD failed - cannot find vnfd-id %s", vnfd_id
)
4203 raise VnfDescriptorError("Cannot find %s", vnfd_id
)
4205 del self
._vnfds
[vnfd_id
]
4207 def nsd_in_use(self
, nsd_id
):
4208 """ Is the NSD with the passed id in use """
4209 self
._log
.debug("Is this NSD in use - msg:%s", nsd_id
)
4210 if nsd_id
in self
._nsds
:
4211 return self
._nsds
[nsd_id
].in_use()
4215 def publish_nsr(self
, xact
, path
, msg
):
4216 """ Publish a NSR """
4217 self
._log
.debug("Publish NSR with path %s, msg %s",
4219 yield from self
.nsr_handler
.update(xact
, path
, msg
)
4222 def unpublish_nsr(self
, xact
, path
):
4223 """ Un Publish an NSR """
4224 self
._log
.debug("Publishing delete NSR with path %s", path
)
4225 yield from self
.nsr_handler
.delete(path
, xact
)
4227 def vnfr_is_ready(self
, vnfr_id
):
4228 """ VNFR with the id is ready """
4229 self
._log
.debug("VNFR id %s ready", vnfr_id
)
4230 if vnfr_id
not in self
._vnfds
:
4231 err
= "Did not find VNFR ID with id %s" % vnfr_id
4232 self
._log
.critical("err")
4233 raise VirtualNetworkFunctionRecordError(err
)
4234 self
._vnfrs
[vnfr_id
].is_ready()
4237 def get_nsd_refcount(self
, nsd_id
):
4238 """ Get the nsd_list from this NSM"""
4240 def nsd_refcount_xpath(nsd_id
):
4241 """ xpath for ref count entry """
4242 return (self
._project
.add_project(NsdRefCountDtsHandler
.XPATH
) +
4243 "[rw-nsr:nsd-id-ref = '{}']").format(nsd_id
)
4246 if nsd_id
is None or nsd_id
== "":
4247 for nsd
in self
._nsds
.values():
4248 nsd_msg
= RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount()
4249 nsd_msg
.nsd_id_ref
= nsd
.id
4250 nsd_msg
.instance_ref_count
= nsd
.ref_count
4251 nsd_list
.append((nsd_refcount_xpath(nsd
.id), nsd_msg
))
4252 elif nsd_id
in self
._nsds
:
4253 nsd_msg
= RwNsrYang
.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount()
4254 nsd_msg
.nsd_id_ref
= self
._nsds
[nsd_id
].id
4255 nsd_msg
.instance_ref_count
= self
._nsds
[nsd_id
].ref_count
4256 nsd_list
.append((nsd_refcount_xpath(nsd_id
), nsd_msg
))
4261 def terminate_ns(self
, nsr_id
, xact
):
4263 Terminate network service for the given NSR Id
4266 # Terminate the instances/networks assocaited with this nw service
4267 self
._log
.debug("Terminating the network service %s", nsr_id
)
4269 yield from self
._nsrs
[nsr_id
].terminate()
4270 except Exception as e
:
4271 self
.log
.exception("Failed to terminate NSR[id=%s]", nsr_id
)
4274 yield from self
.nsd_unref_by_nsr_id(nsr_id
)
4276 # Unpublish the NSR record
4277 self
._log
.debug("Unpublishing the network service %s", nsr_id
)
4278 yield from self
._nsrs
[nsr_id
].unpublish(xact
)
4280 # Finaly delete the NS instance from this NS Manager
4281 self
._log
.debug("Deletng the network service %s", nsr_id
)
4282 self
.delete_nsr(nsr_id
)
4285 class NsmRecordsPublisherProxy(object):
4286 """ This class provides a publisher interface that allows plugin objects
4287 to publish NSR/VNFR/VLR"""
4289 def __init__(self
, dts
, log
, loop
, project
, nsr_pub_hdlr
,
4290 vnfr_pub_hdlr
, vlr_pub_hdlr
,):
4294 self
._project
= project
4295 self
._nsr
_pub
_hdlr
= nsr_pub_hdlr
4296 self
._vlr
_pub
_hdlr
= vlr_pub_hdlr
4297 self
._vnfr
_pub
_hdlr
= vnfr_pub_hdlr
4300 def publish_nsr(self
, xact
, nsr
):
4301 """ Publish an NSR """
4302 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4303 return (yield from self
._nsr
_pub
_hdlr
.update(xact
, path
, nsr
))
4306 def unpublish_nsr(self
, xact
, nsr
):
4307 """ Unpublish an NSR """
4308 path
= NetworkServiceRecord
.xpath_from_nsr(nsr
)
4309 return (yield from self
._nsr
_pub
_hdlr
.delete(xact
, path
))
4312 def publish_vnfr(self
, xact
, vnfr
):
4313 """ Publish an VNFR """
4314 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4315 return (yield from self
._vnfr
_pub
_hdlr
.update(xact
, path
, vnfr
))
4318 def unpublish_vnfr(self
, xact
, vnfr
):
4319 """ Unpublish a VNFR """
4320 path
= VirtualNetworkFunctionRecord
.vnfr_xpath(vnfr
)
4321 return (yield from self
._vnfr
_pub
_hdlr
.delete(xact
, path
))
4324 def publish_vlr(self
, xact
, vlr
):
4325 """ Publish a VLR """
4326 path
= self
._project
.add_project(VirtualLinkRecord
.vlr_xpath(vlr
))
4327 return (yield from self
._vlr
_pub
_hdlr
.update(xact
, path
, vlr
))
4330 def unpublish_vlr(self
, xact
, vlr
):
4331 """ Unpublish a VLR """
4332 path
= self
._project
.add_project(VirtualLinkRecord
.vlr_xpath(vlr
))
4333 return (yield from self
._vlr
_pub
_hdlr
.delete(xact
, path
))
4336 class ScalingRpcHandler(mano_dts
.DtsHandler
):
4337 """ The Network service Monitor DTS handler """
4338 SCALE_IN_INPUT_XPATH
= "I,/nsr:exec-scale-in"
4339 SCALE_IN_OUTPUT_XPATH
= "O,/nsr:exec-scale-in"
4341 SCALE_OUT_INPUT_XPATH
= "I,/nsr:exec-scale-out"
4342 SCALE_OUT_OUTPUT_XPATH
= "O,/nsr:exec-scale-out"
4344 ACTION
= Enum('ACTION', 'SCALE_IN SCALE_OUT')
4346 def __init__(self
, log
, dts
, loop
, project
, callback
=None):
4347 super().__init
__(log
, dts
, loop
, project
)
4348 self
.callback
= callback
4349 self
.last_instance_id
= defaultdict(int)
4350 self
._regh
_in
= None
4351 self
._regh
_out
= None
4357 self
._log
.warning("RPC already registered for project {}".
4358 format(self
._project
.name
))
4362 def on_scale_in_prepare(xact_info
, action
, ks_path
, msg
):
4363 assert action
== rwdts
.QueryAction
.RPC
4366 if not self
._project
.rpc_check(msg
, xact_info
=xact_info
):
4370 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_IN
)
4372 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleIn
.from_dict({
4373 "instance_id": msg
.instance_id
,
4374 "project_name": self
._project
.name
,})
4376 xact_info
.respond_xpath(
4377 rwdts
.XactRspCode
.ACK
,
4378 self
.__class
__.SCALE_IN_OUTPUT_XPATH
,
4381 except Exception as e
:
4382 self
.log
.exception(e
)
4383 xact_info
.respond_xpath(
4384 rwdts
.XactRspCode
.NACK
,
4385 self
.__class
__.SCALE_IN_OUTPUT_XPATH
)
4388 def on_scale_out_prepare(xact_info
, action
, ks_path
, msg
):
4389 assert action
== rwdts
.QueryAction
.RPC
4392 if not self
._project
.rpc_check(msg
, xact_info
=xact_info
):
4395 scaling_group
= msg
.scaling_group_name_ref
4396 if not msg
.instance_id
:
4397 last_instance_id
= self
.last_instance_id
[scale_group
]
4398 msg
.instance_id
= last_instance_id
+ 1
4399 self
.last_instance_id
[scale_group
] += 1
4402 self
.callback(xact_info
.xact
, msg
, self
.ACTION
.SCALE_OUT
)
4404 rpc_op
= NsrYang
.YangOutput_Nsr_ExecScaleOut
.from_dict({
4405 "instance_id": msg
.instance_id
,
4406 "project_name": self
._project
.name
,})
4408 xact_info
.respond_xpath(
4409 rwdts
.XactRspCode
.ACK
,
4410 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
,
4413 except Exception as e
:
4414 self
.log
.exception(e
)
4415 xact_info
.respond_xpath(
4416 rwdts
.XactRspCode
.NACK
,
4417 self
.__class
__.SCALE_OUT_OUTPUT_XPATH
)
4419 scale_in_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4420 on_prepare
=on_scale_in_prepare
)
4421 scale_out_hdl
= rift
.tasklets
.DTS
.RegistrationHandler(
4422 on_prepare
=on_scale_out_prepare
)
4424 with self
.dts
.group_create() as group
:
4425 self
._regh
_in
= group
.register(
4426 xpath
=self
.__class
__.SCALE_IN_INPUT_XPATH
,
4427 handler
=scale_in_hdl
,
4428 flags
=rwdts
.Flag
.PUBLISHER
)
4429 self
._regh
_out
= group
.register(
4430 xpath
=self
.__class
__.SCALE_OUT_INPUT_XPATH
,
4431 handler
=scale_out_hdl
,
4432 flags
=rwdts
.Flag
.PUBLISHER
)
4434 def deregister(self
):
4435 self
._log
.debug("De-register scale RPCs for project {}".
4436 format(self
._project
.name
))
4438 self
._regh
_in
.deregister()
4439 self
._regh
_in
= None
4441 self
._regh
_out
.deregister()
4442 self
._regh
_out
= None
4445 class NsmProject(ManoProject
):
4447 def __init__(self
, name
, tasklet
, **kw
):
4448 super(NsmProject
, self
).__init
__(tasklet
.log
, name
)
4449 self
.update(tasklet
)
4453 self
._ro
_plugin
_selector
= None
4454 self
._vnffgmgr
= None
4456 self
._nsr
_pub
_handler
= None
4457 self
._vnfr
_pub
_handler
= None
4458 self
._vlr
_pub
_handler
= None
4459 self
._vnfd
_pub
_handler
= None
4460 self
._scale
_cfg
_handler
= None
4462 self
._records
_publisher
_proxy
= None
4466 self
._nsr
_pub
_handler
= publisher
.NsrOpDataDtsHandler(
4467 self
._dts
, self
.log
, self
.loop
, self
)
4468 yield from self
._nsr
_pub
_handler
.register()
4470 self
._vnfr
_pub
_handler
= publisher
.VnfrPublisherDtsHandler(
4471 self
._dts
, self
.log
, self
.loop
, self
)
4472 yield from self
._vnfr
_pub
_handler
.register()
4474 self
._vlr
_pub
_handler
= publisher
.VlrPublisherDtsHandler(
4475 self
._dts
, self
.log
, self
.loop
, self
)
4476 yield from self
._vlr
_pub
_handler
.register()
4478 manifest
= self
._tasklet
.tasklet_info
.get_pb_manifest()
4479 use_ssl
= manifest
.bootstrap_phase
.rwsecurity
.use_ssl
4480 ssl_cert
= manifest
.bootstrap_phase
.rwsecurity
.cert
4481 ssl_key
= manifest
.bootstrap_phase
.rwsecurity
.key
4483 self
._vnfd
_pub
_handler
= publisher
.VnfdPublisher(
4484 use_ssl
, ssl_cert
, ssl_key
, self
.loop
, self
)
4486 self
._records
_publisher
_proxy
= NsmRecordsPublisherProxy(
4491 self
._nsr
_pub
_handler
,
4492 self
._vnfr
_pub
_handler
,
4493 self
._vlr
_pub
_handler
,
4496 # Register the NSM to receive the nsm plugin
4497 # when cloud account is configured
4498 self
._ro
_plugin
_selector
= cloud
.ROAccountPluginSelector(
4503 self
._records
_publisher
_proxy
,
4505 yield from self
._ro
_plugin
_selector
.register()
4507 self
._cloud
_account
_handler
= cloud
.CloudAccountConfigSubscriber(
4514 yield from self
._cloud
_account
_handler
.register()
4516 self
._vnffgmgr
= rwvnffgmgr
.VnffgMgr(self
._dts
, self
.log
, self
.log_hdl
, self
.loop
, self
)
4517 yield from self
._vnffgmgr
.register()
4519 self
._nsm
= NsManager(
4524 self
._nsr
_pub
_handler
,
4525 self
._vnfr
_pub
_handler
,
4526 self
._vlr
_pub
_handler
,
4527 self
._ro
_plugin
_selector
,
4529 self
._vnfd
_pub
_handler
,
4530 self
._cloud
_account
_handler
,
4533 yield from self
._nsm
.register()
4535 def deregister(self
):
4536 self
._log
.debug("Project {} de-register".format(self
.name
))
4537 self
._nsm
.deregister()
4538 self
._vnffgmgr
.deregister()
4539 self
._cloud
_account
_handler
.deregister()
4540 self
._ro
_plugin
_selector
.deregister()
4543 class NsmTasklet(rift
.tasklets
.Tasklet
):
4545 The network service manager tasklet
4547 def __init__(self
, *args
, **kwargs
):
4548 super(NsmTasklet
, self
).__init
__(*args
, **kwargs
)
4549 self
.rwlog
.set_category("rw-mano-log")
4550 self
.rwlog
.set_subcategory("nsm")
4553 self
.project_handler
= None
4561 """ The task start callback """
4562 super(NsmTasklet
, self
).start()
4563 self
.log
.info("Starting NsmTasklet")
4565 self
.log
.debug("Registering with dts")
4566 self
._dts
= rift
.tasklets
.DTS(self
.tasklet_info
,
4567 RwNsmYang
.get_schema(),
4569 self
.on_dts_state_change
)
4571 self
.log
.debug("Created DTS Api GI Object: %s", self
._dts
)
4577 print("Caught Exception in NSM stop:", sys
.exc_info()[0])
4580 def on_instance_started(self
):
4581 """ Task instance started callback """
4582 self
.log
.debug("Got instance started callback")
4586 """ Task init callback """
4587 self
.log
.debug("Got instance started callback")
4589 self
.log
.debug("creating project handler")
4590 self
.project_handler
= ProjectHandler(self
, NsmProject
)
4591 self
.project_handler
.register()
4597 """ Task run callback """
4601 def on_dts_state_change(self
, state
):
4602 """Take action according to current dts state to transition
4603 application into the corresponding application state
4606 state - current dts state
4609 rwdts
.State
.INIT
: rwdts
.State
.REGN_COMPLETE
,
4610 rwdts
.State
.CONFIG
: rwdts
.State
.RUN
,
4614 rwdts
.State
.INIT
: self
.init
,
4615 rwdts
.State
.RUN
: self
.run
,
4618 # Transition application to next state
4619 handler
= handlers
.get(state
, None)
4620 if handler
is not None:
4621 yield from handler()
4623 # Transition dts to next state
4624 next_state
= switch
.get(state
, None)
4625 if next_state
is not None:
4626 self
.log
.debug("Changing state to %s", next_state
)
4627 self
._dts
.handle
.set_state(next_state
)