Switch create/delete interfaces to pyroute2

Create and delete network interfaces in ip_lib.py module
now uses pyroute2 library.
Only exception is creation of veth currently as there is no
way to create veth pair and put one end of such veth
in another namespace in one call.

Related-Bug: #1492714

Change-Id: I0a658d91f173fb705b5987a1174bda6a6570468d
This commit is contained in:
Sławek Kapłoński 2018-02-28 16:30:43 +01:00
parent 0c150460c4
commit 53054ad463
4 changed files with 195 additions and 78 deletions

View File

@ -167,10 +167,14 @@ class IPWrapper(SubProcessBase):
return IPDevice(devices[0]['name'], namespace=self.namespace)
def add_tuntap(self, name, mode='tap'):
self._as_root([], 'tuntap', ('add', name, 'mode', mode))
privileged.create_interface(
name, self.namespace, "tuntap", mode=mode)
return IPDevice(name, namespace=self.namespace)
def add_veth(self, name1, name2, namespace2=None):
# TODO(slaweq): switch to pyroute2 when issue
# https://github.com/svinota/pyroute2/issues/463
# will be closed
args = ['add', name1, 'type', 'veth', 'peer', 'name', name2]
if namespace2 is None:
@ -185,18 +189,20 @@ class IPWrapper(SubProcessBase):
IPDevice(name2, namespace=namespace2))
def add_macvtap(self, name, src_dev, mode='bridge'):
args = ['add', 'link', src_dev, 'name', name, 'type', 'macvtap',
'mode', mode]
self._as_root([], 'link', tuple(args))
privileged.create_interface(name,
self.namespace,
"macvtap",
physical_interface=src_dev,
mode=mode)
return IPDevice(name, namespace=self.namespace)
def del_veth(self, name):
"""Delete a virtual interface between two namespaces."""
self._as_root([], 'link', ('del', name))
privileged.delete_interface(name, self.namespace)
def add_dummy(self, name):
"""Create a Linux dummy interface with the given name."""
self._as_root([], 'link', ('add', name, 'type', 'dummy'))
privileged.create_interface(name, self.namespace, "dummy")
return IPDevice(name, namespace=self.namespace)
def ensure_namespace(self, name):
@ -224,35 +230,37 @@ class IPWrapper(SubProcessBase):
device.link.set_netns(self.namespace)
def add_vlan(self, name, physical_interface, vlan_id):
cmd = ['add', 'link', physical_interface, 'name', name,
'type', 'vlan', 'id', vlan_id]
self._as_root([], 'link', cmd)
privileged.create_interface(name,
self.namespace,
"vlan",
physical_interface=physical_interface,
vlan_id=vlan_id)
return IPDevice(name, namespace=self.namespace)
def add_vxlan(self, name, vni, group=None, dev=None, ttl=None, tos=None,
local=None, srcport=None, dstport=None, proxy=False):
cmd = ['add', name, 'type', 'vxlan', 'id', vni]
kwargs = {'vxlan_id': vni}
if group:
cmd.extend(['group', group])
kwargs['vxlan_group'] = group
if dev:
cmd.extend(['dev', dev])
kwargs['physical_interface'] = dev
if ttl:
cmd.extend(['ttl', ttl])
kwargs['vxlan_ttl'] = ttl
if tos:
cmd.extend(['tos', tos])
kwargs['vxlan_tos'] = tos
if local:
cmd.extend(['local', local])
kwargs['vxlan_local'] = local
if proxy:
cmd.append('proxy')
kwargs['vxlan_proxy'] = proxy
# tuple: min,max
if srcport:
if len(srcport) == 2 and srcport[0] <= srcport[1]:
cmd.extend(['srcport', str(srcport[0]), str(srcport[1])])
kwargs['vxlan_port_range'] = (str(srcport[0]), str(srcport[1]))
else:
raise n_exc.NetworkVxlanPortRangeError(vxlan_range=srcport)
if dstport:
cmd.extend(['dstport', str(dstport)])
self._as_root([], 'link', cmd)
kwargs['vxlan_port'] = str(dstport)
privileged.create_interface(name, self.namespace, "vxlan", **kwargs)
return (IPDevice(name, namespace=self.namespace))
@removals.remove(version='Queens', removal_version='Rocky',

View File

@ -13,6 +13,7 @@
import errno
import socket
from neutron_lib import constants
import pyroute2
from pyroute2.netlink import rtnl
from pyroute2.netlink.rtnl import ndmsg
@ -90,16 +91,33 @@ def _get_iproute(namespace):
return pyroute2.IPRoute()
def _run_iproute_neigh(command, device, namespace, **kwargs):
def _get_link_id(device, namespace):
try:
with _get_iproute(namespace) as ip:
idx = ip.link_lookup(ifname=device)[0]
return ip.neigh(command, ifindex=idx, **kwargs)
return ip.link_lookup(ifname=device)[0]
except IndexError:
msg = _("Network interface %(device)s not found in namespace "
"%(namespace)s.") % {'device': device,
'namespace': namespace}
raise NetworkInterfaceNotFound(msg)
def _run_iproute_link(command, device, namespace, **kwargs):
try:
with _get_iproute(namespace) as ip:
idx = _get_link_id(device, namespace)
return ip.link(command, index=idx, **kwargs)
except OSError as e:
if e.errno == errno.ENOENT:
raise NetworkNamespaceNotFound(netns_name=namespace)
raise
def _run_iproute_neigh(command, device, namespace, **kwargs):
try:
with _get_iproute(namespace) as ip:
idx = _get_link_id(device, namespace)
return ip.neigh(command, ifindex=idx, **kwargs)
except OSError as e:
if e.errno == errno.ENOENT:
raise NetworkNamespaceNotFound(netns_name=namespace)
@ -109,13 +127,8 @@ def _run_iproute_neigh(command, device, namespace, **kwargs):
def _run_iproute_addr(command, device, namespace, **kwargs):
try:
with _get_iproute(namespace) as ip:
idx = ip.link_lookup(ifname=device)[0]
idx = _get_link_id(device, namespace)
return ip.addr(command, index=idx, **kwargs)
except IndexError:
msg = _("Network interface %(device)s not found in namespace "
"%(namespace)s.") % {'device': device,
'namespace': namespace}
raise NetworkInterfaceNotFound(msg)
except OSError as e:
if e.errno == errno.ENOENT:
raise NetworkNamespaceNotFound(netns_name=namespace)
@ -152,19 +165,35 @@ def flush_ip_addresses(ip_version, device, namespace):
family = _IP_VERSION_FAMILY_MAP[ip_version]
try:
with _get_iproute(namespace) as ip:
idx = ip.link_lookup(ifname=device)[0]
idx = _get_link_id(device, namespace)
ip.flush_addr(index=idx, family=family)
except IndexError:
msg = _("Network interface %(device)s not found in namespace "
"%(namespace)s.") % {'device': device,
'namespace': namespace}
raise NetworkInterfaceNotFound(msg)
except OSError as e:
if e.errno == errno.ENOENT:
raise NetworkNamespaceNotFound(netns_name=namespace)
raise
@privileged.default.entrypoint
def create_interface(ifname, namespace, kind, **kwargs):
ifname = ifname[:constants.DEVICE_NAME_MAX_LEN]
try:
with _get_iproute(namespace) as ip:
physical_interface = kwargs.pop("physical_interface", None)
if physical_interface:
link_key = "vxlan_link" if kind == "vxlan" else "link"
kwargs[link_key] = _get_link_id(physical_interface, namespace)
return ip.link("add", ifname=ifname, kind=kind, **kwargs)
except OSError as e:
if e.errno == errno.ENOENT:
raise NetworkNamespaceNotFound(netns_name=namespace)
raise
@privileged.default.entrypoint
def delete_interface(ifname, namespace, **kwargs):
_run_iproute_link("del", ifname, namespace, **kwargs)
@privileged.default.entrypoint
def add_neigh_entry(ip_version, ip_address, mac_address, device, namespace,
**kwargs):

View File

@ -341,11 +341,10 @@ class TestIpWrapper(base.BaseTestCase):
self.assertEqual(1, priv_listnetns.call_count)
self.assertFalse(listnetns.called)
def test_add_tuntap(self):
@mock.patch.object(priv_lib, 'create_interface')
def test_add_tuntap(self, create):
ip_lib.IPWrapper().add_tuntap('tap0')
self.execute.assert_called_once_with([], 'tuntap',
('add', 'tap0', 'mode', 'tap'),
run_as_root=True, namespace=None)
create.assert_called_once_with('tap0', None, 'tuntap', mode='tap')
def test_add_veth(self):
ip_lib.IPWrapper().add_veth('tap0', 'tap1')
@ -354,19 +353,17 @@ class TestIpWrapper(base.BaseTestCase):
'peer', 'name', 'tap1'),
run_as_root=True, namespace=None)
def test_add_macvtap(self):
@mock.patch.object(priv_lib, 'create_interface')
def test_add_macvtap(self, create):
ip_lib.IPWrapper().add_macvtap('macvtap0', 'eth0', 'bridge')
self.execute.assert_called_once_with([], 'link',
('add', 'link', 'eth0', 'name',
'macvtap0', 'type', 'macvtap',
'mode', 'bridge'),
run_as_root=True, namespace=None)
create.assert_called_once_with(
'macvtap0', None, 'macvtap', physical_interface='eth0',
mode='bridge')
def test_del_veth(self):
@mock.patch.object(priv_lib, 'delete_interface')
def test_del_veth(self, delete):
ip_lib.IPWrapper().del_veth('fpr-1234')
self.execute.assert_called_once_with([], 'link',
('del', 'fpr-1234'),
run_as_root=True, namespace=None)
delete.assert_called_once_with('fpr-1234', None)
def test_add_veth_with_namespaces(self):
ns2 = 'ns2'
@ -379,12 +376,10 @@ class TestIpWrapper(base.BaseTestCase):
'netns', ns2),
run_as_root=True, namespace=None)
def test_add_dummy(self):
@mock.patch.object(priv_lib, 'create_interface')
def test_add_dummy(self, create):
ip_lib.IPWrapper().add_dummy('dummy0')
self.execute.assert_called_once_with([], 'link',
('add', 'dummy0',
'type', 'dummy'),
run_as_root=True, namespace=None)
create.assert_called_once_with('dummy0', None, 'dummy')
def test_get_device(self):
dev = ip_lib.IPWrapper(namespace='ns').device('eth0')
@ -477,17 +472,42 @@ class TestIpWrapper(base.BaseTestCase):
self.assertNotIn(mock.call().delete('ns'),
ip_ns_cmd_cls.mock_calls)
def test_add_vlan(self):
@mock.patch.object(priv_lib, 'create_interface')
def test_add_vlan(self, create):
retval = ip_lib.IPWrapper().add_vlan('eth0.1', 'eth0', '1')
self.assertIsInstance(retval, ip_lib.IPDevice)
self.assertEqual(retval.name, 'eth0.1')
self.execute.assert_called_once_with([], 'link',
['add', 'link', 'eth0',
'name', 'eth0.1',
'type', 'vlan', 'id', '1'],
run_as_root=True, namespace=None)
create.assert_called_once_with('eth0.1',
None,
'vlan',
physical_interface='eth0',
vlan_id='1')
@mock.patch.object(priv_lib, 'create_interface')
def test_add_vxlan_valid_srcport_length(self, create):
self.call_params = {}
def fake_create_interface(ifname, namespace, kind, **kwargs):
self.call_params = dict(
ifname=ifname,
namespace=namespace,
kind=kind,
**kwargs)
create.side_effect = fake_create_interface
expected_call_params = {
'ifname': 'vxlan0',
'namespace': None,
'kind': 'vxlan',
'vxlan_id': 'vni0',
'vxlan_group': 'group0',
'physical_interface': 'dev0',
'vxlan_ttl': 'ttl0',
'vxlan_tos': 'tos0',
'vxlan_local': 'local0',
'vxlan_proxy': True,
'vxlan_port_range': ('1', '2')}
def test_add_vxlan_valid_srcport_length(self):
retval = ip_lib.IPWrapper().add_vxlan('vxlan0', 'vni0',
group='group0',
dev='dev0', ttl='ttl0',
@ -496,14 +516,7 @@ class TestIpWrapper(base.BaseTestCase):
srcport=(1, 2))
self.assertIsInstance(retval, ip_lib.IPDevice)
self.assertEqual(retval.name, 'vxlan0')
self.execute.assert_called_once_with([], 'link',
['add', 'vxlan0', 'type',
'vxlan', 'id', 'vni0', 'group',
'group0', 'dev', 'dev0',
'ttl', 'ttl0', 'tos', 'tos0',
'local', 'local0', 'proxy',
'srcport', '1', '2'],
run_as_root=True, namespace=None)
self.assertDictEqual(expected_call_params, self.call_params)
def test_add_vxlan_invalid_srcport_length(self):
wrapper = ip_lib.IPWrapper()
@ -521,7 +534,32 @@ class TestIpWrapper(base.BaseTestCase):
local='local0', proxy=True,
srcport=(2000, 1000))
def test_add_vxlan_dstport(self):
@mock.patch.object(priv_lib, 'create_interface')
def test_add_vxlan_dstport(self, create):
self.call_params = {}
def fake_create_interface(ifname, namespace, kind, **kwargs):
self.call_params = dict(
ifname=ifname,
namespace=namespace,
kind=kind,
**kwargs)
create.side_effect = fake_create_interface
expected_call_params = {
'ifname': 'vxlan0',
'namespace': None,
'kind': 'vxlan',
'vxlan_id': 'vni0',
'vxlan_group': 'group0',
'physical_interface': 'dev0',
'vxlan_ttl': 'ttl0',
'vxlan_tos': 'tos0',
'vxlan_local': 'local0',
'vxlan_proxy': True,
'vxlan_port_range': ('1', '2'),
'vxlan_port': '4789'}
retval = ip_lib.IPWrapper().add_vxlan('vxlan0', 'vni0',
group='group0',
dev='dev0', ttl='ttl0',
@ -532,15 +570,7 @@ class TestIpWrapper(base.BaseTestCase):
self.assertIsInstance(retval, ip_lib.IPDevice)
self.assertEqual(retval.name, 'vxlan0')
self.execute.assert_called_once_with([], 'link',
['add', 'vxlan0', 'type',
'vxlan', 'id', 'vni0', 'group',
'group0', 'dev', 'dev0',
'ttl', 'ttl0', 'tos', 'tos0',
'local', 'local0', 'proxy',
'srcport', '1', '2',
'dstport', '4789'],
run_as_root=True, namespace=None)
self.assertDictEqual(expected_call_params, self.call_params)
def test_add_device_to_namespace(self):
dev = mock.Mock()

View File

@ -23,6 +23,54 @@ from neutron.tests import base
class IpLibTestCase(base.BaseTestCase):
def _test_run_iproute_link(self, namespace=None):
ip_obj = "NetNS" if namespace else "IPRoute"
with mock.patch.object(pyroute2, ip_obj) as ip_mock_cls:
ip_mock = ip_mock_cls()
ip_mock.__enter__().link_lookup.return_value = [2]
priv_lib._run_iproute_link("test_cmd", "eth0", namespace,
test_param="test_value")
ip_mock.assert_has_calls([
mock.call.__enter__().link_lookup(ifname="eth0"),
mock.call.__exit__(None, None, None),
mock.call.__enter__().link("test_cmd", index=2,
test_param="test_value")])
def test_run_iproute_link_no_namespace(self):
self._test_run_iproute_link()
def test_run_iproute_link_in_namespace(self):
self._test_run_iproute_link(namespace="testns")
def test_run_iproute_link_interface_not_exists(self):
with mock.patch.object(pyroute2, "IPRoute") as iproute_mock:
ip_mock = iproute_mock()
ip_mock.__enter__().link_lookup.return_value = []
self.assertRaises(
priv_lib.NetworkInterfaceNotFound,
priv_lib._run_iproute_link,
"test_cmd", "eth0", None, test_param="test_value")
def test_run_iproute_link_namespace_not_exists(self):
with mock.patch.object(pyroute2, "IPRoute") as iproute_mock:
iproute_mock.side_effect = OSError(
errno.ENOENT, "Test no netns exception")
self.assertRaises(
priv_lib.NetworkNamespaceNotFound,
priv_lib._run_iproute_link,
"test_cmd", "eth0", None, test_param="test_value")
def test_run_iproute_link_error(self):
with mock.patch.object(pyroute2, "IPRoute") as iproute_mock:
iproute_mock.side_effect = OSError(
errno.EINVAL, "Test invalid argument exception")
try:
priv_lib._run_iproute_link(
"test_cmd", "eth0", None, test_param="test_value")
self.fail("OSError exception not raised")
except OSError as e:
self.assertEqual(errno.EINVAL, e.errno)
def _test_run_iproute_neigh(self, namespace=None):
ip_obj = "NetNS" if namespace else "IPRoute"
with mock.patch.object(pyroute2, ip_obj) as ip_mock_cls:
@ -32,6 +80,7 @@ class IpLibTestCase(base.BaseTestCase):
test_param="test_value")
ip_mock.assert_has_calls([
mock.call.__enter__().link_lookup(ifname="eth0"),
mock.call.__exit__(None, None, None),
mock.call.__enter__().neigh("test_cmd", ifindex=2,
test_param="test_value")])
@ -79,6 +128,7 @@ class IpLibTestCase(base.BaseTestCase):
test_param="test_value")
ip_mock.assert_has_calls([
mock.call.__enter__().link_lookup(ifname="eth0"),
mock.call.__exit__(None, None, None),
mock.call.__enter__().addr("test_cmd", index=2,
test_param="test_value")])