1 ##############################################################################
2 # Copyright (c) 2016 Feng Pan (fpan@redhat.com) and others.
4 # All rights reserved. This program and the accompanying materials
5 # are made available under the terms of the Apache License, Version 2.0
6 # which accompanies this distribution, and is available at
7 # http://www.apache.org/licenses/LICENSE-2.0
8 ##############################################################################
13 from . import ip_utils
14 from .common import constants, utils
17 class NetworkSettings:
19 This class parses APEX network settings yaml file into an object. It
20 generates or detects all missing fields for deployment.
22 The resulting object will be used later to generate network environment file
23 as well as configuring post deployment networks.
25 Currently the parsed object is dumped into a bash global definition file
26 for deploy.sh consumption. This object will later be used directly as
27 deployment script move to python.
29 def __init__(self, filename, network_isolation):
30 with open(filename, 'r') as network_settings_file:
31 self.settings_obj = yaml.load(network_settings_file)
32 self.network_isolation = network_isolation
33 self.enabled_network_list = []
34 self._validate_input()
36 def _validate_input(self):
38 Validates the network settings file and populates all fields.
40 NetworkSettingsException will be raised if validation fails.
42 if constants.ADMIN_NETWORK not in self.settings_obj or \
43 not utils.str2bool(self.settings_obj[constants.ADMIN_NETWORK].get(
45 raise NetworkSettingsException("You must enable admin_network "
46 "and configure it explicitly or "
48 if self.network_isolation and \
49 (constants.PUBLIC_NETWORK not in self.settings_obj or not
50 utils.str2bool(self.settings_obj[constants.PUBLIC_NETWORK].get(
52 raise NetworkSettingsException("You must enable public_network "
53 "and configure it explicitly or "
56 for network in constants.OPNFV_NETWORK_TYPES:
57 if network in self.settings_obj:
58 if utils.str2bool(self.settings_obj[network].get('enabled')):
59 logging.info("{} enabled".format(network))
60 self._config_required_settings(network)
61 self._config_ip_range(network=network,
62 setting='usable_ip_range',
63 start_offset=21, end_offset=21)
64 self._config_optional_settings(network)
65 self.enabled_network_list.append(network)
67 logging.info("{} disabled, will collapse with "
68 "admin_network".format(network))
70 logging.info("{} is not in specified, will collapse with "
71 "admin_network".format(network))
73 self.settings_obj['dns_servers'] = self.settings_obj.get(
74 'dns_servers', constants.DNS_SERVERS)
76 def _config_required_settings(self, network):
78 Configures either CIDR or bridged_interface setting
80 cidr takes precedence if both cidr and bridged_interface are specified
83 When using bridged_interface, we will detect network setting on the
84 given NIC in the system. The resulting config in settings object will
85 be an ipaddress.network object, replacing the NIC name.
87 # if vlan not defined then default it to native
88 if network is not constants.ADMIN_NETWORK:
89 if 'vlan' not in self.settings_obj[network]:
90 self.settings_obj[network]['vlan'] = 'native'
92 cidr = self.settings_obj[network].get('cidr')
93 nic_name = self.settings_obj[network].get('bridged_interface')
96 cidr = ipaddress.ip_network(self.settings_obj[network]['cidr'])
97 self.settings_obj[network]['cidr'] = cidr
98 logging.info("{}_cidr: {}".format(network, cidr))
101 # If cidr is not specified, we need to know if we should find
102 # IPv6 or IPv4 address on the interface
103 if utils.str2bool(self.settings_obj[network].get('ipv6')):
107 nic_interface = ip_utils.get_interface(nic_name, address_family)
109 self.settings_obj[network]['bridged_interface'] = nic_interface
110 logging.info("{}_bridged_interface: {}".
111 format(network, nic_interface))
114 raise NetworkSettingsException("Auto detection failed for {}: "
115 "Unable to find valid ip for "
117 .format(network, nic_name))
120 raise NetworkSettingsException("Auto detection failed for {}: "
121 "either bridge_interface or cidr "
125 def _config_ip_range(self, network, setting, start_offset=None,
126 end_offset=None, count=None):
128 Configures IP range for a given setting.
130 If the setting is already specified, no change will be made.
132 The spec for start_offset, end_offset and count are identical to
133 ip_utils.get_ip_range.
135 ip_range = self.settings_obj[network].get(setting)
136 interface = self.settings_obj[network].get('bridged_interface')
139 cidr = self.settings_obj[network].get('cidr')
140 ip_range = ip_utils.get_ip_range(start_offset=start_offset,
141 end_offset=end_offset,
145 self.settings_obj[network][setting] = ip_range
147 logging.info("{}_{}: {}".format(network, setting, ip_range))
149 def _config_ip(self, network, setting, offset):
151 Configures IP for a given setting.
153 If the setting is already specified, no change will be made.
155 The spec for offset is identical to ip_utils.get_ip
157 ip = self.settings_obj[network].get(setting)
158 interface = self.settings_obj[network].get('bridged_interface')
161 cidr = self.settings_obj[network].get('cidr')
162 ip = ip_utils.get_ip(offset, cidr, interface)
163 self.settings_obj[network][setting] = ip
165 logging.info("{}_{}: {}".format(network, setting, ip))
167 def _config_optional_settings(self, network):
169 Configures optional settings:
173 - introspection_range
179 if network == constants.ADMIN_NETWORK:
180 self._config_ip(network, 'provisioner_ip', 1)
181 self._config_ip_range(network=network, setting='dhcp_range',
182 start_offset=2, count=9)
183 self._config_ip_range(network=network,
184 setting='introspection_range',
185 start_offset=11, count=9)
186 elif network == constants.PUBLIC_NETWORK:
187 self._config_ip(network, 'provisioner_ip', 1)
188 self._config_ip_range(network=network,
189 setting='floating_ip',
190 end_offset=2, count=20)
191 self._config_gateway(network)
193 def _config_gateway(self, network):
195 Configures gateway setting for a given network.
197 If cidr is specified, we always use the first address in the address
198 space for gateway. Otherwise, we detect the system gateway.
200 gateway = self.settings_obj[network].get('gateway')
201 interface = self.settings_obj[network].get('bridged_interface')
204 cidr = self.settings_obj[network].get('cidr')
206 gateway = ip_utils.get_ip(1, cidr)
208 gateway = ip_utils.find_gateway(interface)
211 self.settings_obj[network]['gateway'] = gateway
213 raise NetworkSettingsException("Failed to set gateway")
215 logging.info("{}_gateway: {}".format(network, gateway))
217 def dump_bash(self, path=None):
219 Prints settings for bash consumption.
221 If optional path is provided, bash string will be written to the file
225 for network in self.enabled_network_list:
226 for key, value in self.settings_obj[network].items():
227 bash_str += "{}_{}={}\n".format(network, key, value)
228 bash_str += "enabled_network_list='{}'\n" \
229 .format(' '.join(self.enabled_network_list))
230 bash_str += "ip_addr_family={}\n".format(self.get_ip_addr_family())
232 for dns_server in self.settings_obj['dns_servers']:
233 dns_list = dns_list + "{} ".format(dns_server)
234 dns_list = dns_list.strip()
235 bash_str += "dns_servers=\'{}\'\n".format(dns_list)
237 with open(path, 'w') as file:
242 def get_ip_addr_family(self):
244 Returns IP address family for current deployment.
246 If any enabled network has IPv6 CIDR, the deployment is classified as
249 for network in self.enabled_network_list:
250 cidr = ipaddress.ip_network(self.settings_obj[network]['cidr'])
251 if cidr.version == 6:
256 def get_network_settings(self):
258 Getter for network settings
259 :return: network settings dictionary
261 return self.settings_obj
263 def get_enabled_networks(self):
265 Getter for enabled network list
266 :return: list of enabled networks
268 return self.enabled_network_list
271 class NetworkSettingsException(Exception):
272 def __init__(self, value):