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 cidr = self.settings_obj[network].get('cidr')
88 nic_name = self.settings_obj[network].get('bridged_interface')
91 cidr = ipaddress.ip_network(self.settings_obj[network]['cidr'])
92 self.settings_obj[network]['cidr'] = cidr
93 logging.info("{}_cidr: {}".format(network, cidr))
96 # If cidr is not specified, we need to know if we should find
97 # IPv6 or IPv4 address on the interface
98 if utils.str2bool(self.settings_obj[network].get('ipv6')):
102 nic_interface = ip_utils.get_interface(nic_name, address_family)
104 self.settings_obj[network]['bridged_interface'] = nic_interface
105 logging.info("{}_bridged_interface: {}".
106 format(network, nic_interface))
109 raise NetworkSettingsException("Auto detection failed for {}: "
110 "Unable to find valid ip for "
112 .format(network, nic_name))
115 raise NetworkSettingsException("Auto detection failed for {}: "
116 "either bridge_interface or cidr "
120 def _config_ip_range(self, network, setting, start_offset=None,
121 end_offset=None, count=None):
123 Configures IP range for a given setting.
125 If the setting is already specified, no change will be made.
127 The spec for start_offset, end_offset and count are identical to
128 ip_utils.get_ip_range.
130 ip_range = self.settings_obj[network].get(setting)
131 interface = self.settings_obj[network].get('bridged_interface')
134 cidr = self.settings_obj[network].get('cidr')
135 ip_range = ip_utils.get_ip_range(start_offset=start_offset,
136 end_offset=end_offset,
140 self.settings_obj[network][setting] = ip_range
142 logging.info("{}_{}: {}".format(network, setting, ip_range))
144 def _config_ip(self, network, setting, offset):
146 Configures IP for a given setting.
148 If the setting is already specified, no change will be made.
150 The spec for offset is identical to ip_utils.get_ip
152 ip = self.settings_obj[network].get(setting)
153 interface = self.settings_obj[network].get('bridged_interface')
156 cidr = self.settings_obj[network].get('cidr')
157 ip = ip_utils.get_ip(offset, cidr, interface)
158 self.settings_obj[network][setting] = ip
160 logging.info("{}_{}: {}".format(network, setting, ip))
162 def _config_optional_settings(self, network):
164 Configures optional settings:
168 - introspection_range
174 if network == constants.ADMIN_NETWORK:
175 self._config_ip(network, 'provisioner_ip', 1)
176 self._config_ip_range(network=network, setting='dhcp_range',
177 start_offset=2, count=9)
178 self._config_ip_range(network=network,
179 setting='introspection_range',
180 start_offset=11, count=9)
181 elif network == constants.PUBLIC_NETWORK:
182 self._config_ip(network, 'provisioner_ip', 1)
183 self._config_ip_range(network=network,
184 setting='floating_ip',
185 end_offset=2, count=20)
186 self._config_gateway(network)
188 def _config_gateway(self, network):
190 Configures gateway setting for a given network.
192 If cidr is specified, we always use the first address in the address
193 space for gateway. Otherwise, we detect the system gateway.
195 gateway = self.settings_obj[network].get('gateway')
196 interface = self.settings_obj[network].get('bridged_interface')
199 cidr = self.settings_obj[network].get('cidr')
201 gateway = ip_utils.get_ip(1, cidr)
203 gateway = ip_utils.find_gateway(interface)
206 self.settings_obj[network]['gateway'] = gateway
208 raise NetworkSettingsException("Failed to set gateway")
210 logging.info("{}_gateway: {}".format(network, gateway))
212 def dump_bash(self, path=None):
214 Prints settings for bash consumption.
216 If optional path is provided, bash string will be written to the file
220 for network in self.enabled_network_list:
221 for key, value in self.settings_obj[network].items():
222 bash_str += "{}_{}={}\n".format(network, key, value)
223 bash_str += "enabled_network_list='{}'\n" \
224 .format(' '.join(self.enabled_network_list))
225 bash_str += "ip_addr_family={}\n".format(self.get_ip_addr_family())
227 for dns_server in self.settings_obj['dns_servers']:
228 dns_list = dns_list + "{} ".format(dns_server)
229 dns_list = dns_list.strip()
230 bash_str += "dns_servers=\'{}\'\n".format(dns_list)
232 with open(path, 'w') as file:
237 def get_ip_addr_family(self):
239 Returns IP address family for current deployment.
241 If any enabled network has IPv6 CIDR, the deployment is classified as
244 for network in self.enabled_network_list:
245 cidr = ipaddress.ip_network(self.settings_obj[network]['cidr'])
246 if cidr.version == 6:
251 def get_network_settings(self):
253 Getter for network settings
254 :return: network settings dictionary
256 return self.settings_obj
258 def get_enabled_networks(self):
260 Getter for enabled network list
261 :return: list of enabled networks
263 return self.enabled_network_list
266 class NetworkSettingsException(Exception):
267 def __init__(self, value):