import ipaddress
import mysql.connector
import requests
-import commands
+import subprocess
import time
import yaml
from novaclient import client as openstackclient
+from openstack.config import loader
from kubernetes import client, config
from netaddr import IPAddress, IPNetwork
class VcpeCommon:
- def __init__(self, extra_host_names=None):
+ def __init__(self, extra_host_names=None, cfg_file=None):
self.logger = logging.getLogger(__name__)
self.logger.setLevel(logging.DEBUG)
self.logger.info('Initializing configuration')
+ self.default_config = 'vcpeconfig.yaml'
# Read configuration from config file
- self._load_config()
+ self._load_config(cfg_file)
+ # Load OpenStack settings
+ self._load_os_config()
self.sdnc_controller_pod = '-'.join([self.onap_environment, 'sdnc-sdnc-0'])
# OOM: this is the address that the brg and bng will nat for sdnc access - 10.0.0.x address of k8 host for sdnc-0 container
'Content-Type': 'application/json',
'X-FromAppId': 'postman', 'X-TransactionId': '9999'}
- def _load_config(self, cfg_file='vcpeconfig.yaml'):
+ def _load_config(self, cfg_file):
"""
Reads vcpe config file and injects settings as object's attributes
:param cfg_file: Configuration file path
"""
+ if cfg_file is None:
+ cfg_file = self.default_config
+
try:
with open(cfg_file, 'r') as cfg:
cfg_yml = yaml.full_load(cfg)
self.logger.error('Unable to parse config file: ' + str(e))
sys.exit(1)
+ def _load_os_config(self):
+ """
+ Reads cloud settings and sets them as object's 'cloud' attribute
+ """
+ # Create OpenStackConfig config instance
+ os_config = loader.OpenStackConfig()
+ # Try reading cloud settings for self.cloud_name
+ try:
+ os_cloud = os_config.cloud_config['clouds'][self.cloud_name]
+ except KeyError:
+ self.logger.error('Error fetching cloud settings for cloud "{0}"'
+ .format(self.cloud_name))
+ sys.exit(1)
+ self.logger.debug('Cloud config:\n {0}'.format(json.dumps(
+ os_cloud,indent=4)))
+
+ # Extract all OS settings keys and alter their names
+ # to conform to openstack cli client
+ self.cloud = {}
+ for k in os_cloud:
+ if isinstance(os_cloud[k],dict):
+ for sub_k in os_cloud[k]:
+ os_setting_name = '--os-' + sub_k.replace('_','-')
+ self.cloud[os_setting_name] = os_cloud[k][sub_k]
+ else:
+ os_setting_name = '--os-' + k.replace('_','-')
+ self.cloud[os_setting_name] = os_cloud[k]
+
def heatbridge(self, openstack_stack_name, svc_instance_uuid):
"""
Add vserver information to AAI
self.logger.info('Adding vServer information to AAI for {0}'.format(openstack_stack_name))
if not self.oom_mode:
cmd = '/opt/demo.sh heatbridge {0} {1} vCPE'.format(openstack_stack_name, svc_instance_uuid)
- ret = commands.getstatusoutput("ssh -i onap_dev root@{0} '{1}'".format(self.hosts['robot'], cmd))
+ ret = subprocess.getstatusoutput("ssh -i onap_dev root@{0} '{1}'".format(self.hosts['robot'], cmd))
self.logger.debug('%s', ret)
else:
print('To add vGMUX vserver info to AAI, do the following:')
:param sz: a string
:return: the first IP address matching the network, e.g. 10.5.12.3
"""
- network = ipaddress.ip_network(unicode('{0}/{1}'.format(net_addr, net_addr_len)), strict=False)
+ network = ipaddress.ip_network(unicode('{0}/{1}'.format(net_addr, net_addr_len)), strict=False) # pylint: disable=E0602
ip_list = re.findall(r'[0-9]+(?:\.[0-9]+){3}', sz)
for ip in ip_list:
- this_net = ipaddress.ip_network(unicode('{0}/{1}'.format(ip, net_addr_len)), strict=False)
+ this_net = ipaddress.ip_network(unicode('{0}/{1}'.format(ip, net_addr_len)), strict=False) # pylint: disable=E0602
if this_net == network:
return str(ip)
return None
def get_pod_node_oam_ip(self, pod):
"""
- :Assuming kubectl is available and configured by default config (~/.kube/config)
+ :Assuming kubectl is available and configured by default config (~/.kube/config)
:param pod: pod name substring, e.g. 'sdnc-sdnc-0'
:return pod's cluster node oam ip (10.0.0.0/16)
"""
break
if ret is None:
- ret = raw_input("Enter " + self.sdnc_controller_pod + " pod cluster node OAM IP address(10.0.0.0/16): ")
+ ret = raw_input("Enter " + self.sdnc_controller_pod + " pod cluster node OAM IP address(10.0.0.0/16): ") # pylint: disable=E0602
return ret
def get_pod_node_public_ip(self, pod):
"""
- :Assuming kubectl is available and configured by default config (~/.kube/config)
+ :Assuming kubectl is available and configured by default config (~/.kube/config)
:param pod: pod name substring, e.g. 'sdnc-sdnc-0'
:return pod's cluster node public ip (i.e. 10.12.0.0/16)
"""
break
if ret is None:
- ret = raw_input("Enter " + self.sdnc_controller_pod + " pod cluster node public IP address(i.e. " + self.external_net_addr + "): ")
+ ret = raw_input("Enter " + self.sdnc_controller_pod + " pod cluster node public IP address(i.e. " + self.external_net_addr + "): ") # pylint: disable=E0602
return ret
def get_vm_public_ip_by_nova(self, vm):
:return vm public ip
"""
subnet = IPNetwork('{0}/{1}'.format(self.external_net_addr, self.external_net_prefix_len))
- nova = openstackclient.Client(2, self.cloud['--os-username'], self.cloud['--os-password'], self.cloud['--os-tenant-id'], self.cloud['--os-auth-url'])
+ nova = openstackclient.Client(2, self.cloud['--os-username'], self.cloud['--os-password'], self.cloud['--os-tenant-id'], self.cloud['--os-auth-url'])
for i in nova.servers.list():
if i.name == vm:
- for k, v in i.networks.items():
+ for k, v in i.networks.items(): # pylint: disable=W0612
for ip in v:
if IPAddress(ip) in subnet:
return ip
url = self.vpp_inf_url.format(ip) + '/interface/' + inf
requests.delete(url, headers=self.vpp_api_headers, auth=self.vpp_api_userpass)
- if len(self.get_vxlan_interfaces(ip)) > 0:
+ if self.get_vxlan_interfaces(ip):
self.logger.error("Error deleting VxLAN from {0}, try to restart the VM, IP is {1}.".format(host, ip))
return False
def load_vgmux_vnf_name(self):
return self.load_object(self.vgmux_vnf_name_file)
-
Code Review / integration.git / blobdiff