LOG = logging.getLogger(__name__)
-def vxlan_supported(root_helper, from_ip='192.0.2.1', to_ip='192.0.2.2'):
+def ovs_vxlan_supported(root_helper, from_ip='192.0.2.1', to_ip='192.0.2.2'):
name = "vxlantest-" + utils.get_random_string(6)
with ovs_lib.OVSBridge(name, root_helper) as br:
port = br.add_tunnel_port(from_ip, to_ip, const.TYPE_VXLAN)
return port != ovs_lib.INVALID_OFPORT
+def iproute2_vxlan_supported(root_helper):
+ ip = ip_lib.IPWrapper(root_helper)
+ name = "vxlantest-" + utils.get_random_string(4)
+ port = ip.add_vxlan(name, 3000)
+ ip.del_veth(name)
+ return name == port.name
+
+
def patch_supported(root_helper):
seed = utils.get_random_string(6)
name = "patchtest-" + seed
LOG = logging.getLogger(__name__)
cfg.CONF.import_group('AGENT', 'neutron.plugins.openvswitch.common.config')
cfg.CONF.import_group('OVS', 'neutron.plugins.openvswitch.common.config')
+cfg.CONF.import_group('VXLAN', 'neutron.plugins.linuxbridge.common.config')
+cfg.CONF.import_group('ml2', 'neutron.plugins.ml2.config')
cfg.CONF.import_group('ml2_sriov',
'neutron.plugins.ml2.drivers.mech_sriov.mech_driver')
def check_ovs_vxlan():
- result = checks.vxlan_supported(root_helper=cfg.CONF.AGENT.root_helper)
+ result = checks.ovs_vxlan_supported(root_helper=cfg.CONF.AGENT.root_helper)
if not result:
LOG.error(_LE('Check for Open vSwitch VXLAN support failed. '
'Please ensure that the version of openvswitch '
return result
+def check_iproute2_vxlan():
+ result = checks.iproute2_vxlan_supported(
+ root_helper=cfg.CONF.AGENT.root_helper)
+ if not result:
+ LOG.error(_LE('Check for iproute2 VXLAN support failed. Please ensure '
+ 'that the iproute2 has VXLAN support.'))
+ return result
+
+
def check_ovs_patch():
result = checks.patch_supported(root_helper=cfg.CONF.AGENT.root_helper)
if not result:
# Define CLI opts to test specific features, with a calback for the test
OPTS = [
- BoolOptCallback('ovs_vxlan', check_ovs_vxlan,
- help=_('Check for vxlan support')),
- BoolOptCallback('ovs_patch', check_ovs_patch,
+ BoolOptCallback('ovs_vxlan', check_ovs_vxlan, default=False,
+ help=_('Check for OVS vxlan support')),
+ BoolOptCallback('iproute2_vxlan', check_iproute2_vxlan, default=False,
+ help=_('Check for iproute2 vxlan support')),
+ BoolOptCallback('ovs_patch', check_ovs_patch, default=False,
help=_('Check for patch port support')),
BoolOptCallback('nova_notify', check_nova_notify,
help=_('Check for nova notification support')),
if 'vxlan' in cfg.CONF.AGENT.tunnel_types:
cfg.CONF.set_override('ovs_vxlan', True)
+ if ('vxlan' in cfg.CONF.ml2.type_drivers or
+ cfg.CONF.VXLAN.enable_vxlan):
+ cfg.CONF.set_override('iproute2_vxlan', True)
if cfg.CONF.AGENT.tunnel_types:
cfg.CONF.set_override('ovs_patch', True)
if not cfg.CONF.OVS.use_veth_interconnection:
return False
return True
- def vxlan_module_supported(self):
- try:
- utils.execute(cmd=['modinfo', 'vxlan'], log_fail_as_error=False)
- return True
- except RuntimeError:
- return False
-
def check_vxlan_support(self):
self.vxlan_mode = lconst.VXLAN_NONE
- if not self.vxlan_module_supported():
- LOG.error(_LE('Linux kernel vxlan module and iproute2 3.8 or '
- 'above are required to enable VXLAN.'))
- raise exceptions.VxlanNetworkUnsupported()
if self.vxlan_ucast_supported():
self.vxlan_mode = lconst.VXLAN_UCAST
self.check_sudo_enabled()
def test_ovs_vxlan_support_runs(self):
- checks.vxlan_supported(self.root_helper)
+ checks.ovs_vxlan_supported(self.root_helper)
+
+ def test_iproute2_vxlan_support_runs(self):
+ checks.iproute2_vxlan_supported(self.root_helper)
def test_ovs_patch_support_runs(self):
checks.patch_supported(self.root_helper)
self.lbm.delete_vlan("eth1.1")
self.assertTrue(exec_fn.called)
- def _check_vxlan_support(self, expected, vxlan_module_supported,
- vxlan_ucast_supported, vxlan_mcast_supported):
+ def _check_vxlan_support(self, expected, vxlan_ucast_supported,
+ vxlan_mcast_supported):
with contextlib.nested(
- mock.patch.object(self.lbm, 'vxlan_module_supported',
- return_value=vxlan_module_supported),
mock.patch.object(self.lbm, 'vxlan_ucast_supported',
return_value=vxlan_ucast_supported),
mock.patch.object(self.lbm, 'vxlan_mcast_supported',
def test_check_vxlan_support(self):
self._check_vxlan_support(expected=lconst.VXLAN_UCAST,
- vxlan_module_supported=True,
vxlan_ucast_supported=True,
vxlan_mcast_supported=True)
self._check_vxlan_support(expected=lconst.VXLAN_MCAST,
- vxlan_module_supported=True,
vxlan_ucast_supported=False,
vxlan_mcast_supported=True)
self._check_vxlan_support(expected=lconst.VXLAN_NONE,
- vxlan_module_supported=False,
vxlan_ucast_supported=False,
vxlan_mcast_supported=False)
self._check_vxlan_support(expected=lconst.VXLAN_NONE,
- vxlan_module_supported=True,
vxlan_ucast_supported=False,
vxlan_mcast_supported=False)
- def _check_vxlan_module_supported(self, expected, execute_side_effect):
- with mock.patch.object(
- utils, 'execute',
- side_effect=execute_side_effect):
- self.assertEqual(expected, self.lbm.vxlan_module_supported())
-
- def test_vxlan_module_supported(self):
- self._check_vxlan_module_supported(
- expected=True,
- execute_side_effect=None)
- self._check_vxlan_module_supported(
- expected=False,
- execute_side_effect=RuntimeError())
-
def _check_vxlan_ucast_supported(
self, expected, l2_population, iproute_arg_supported, fdb_append):
cfg.CONF.set_override('l2_population', l2_population, 'VXLAN')