openstack
/
neutron
Code Issues Proposed changes

Add IPWrapper.get_devices_info using PyRoute2

Browse Source 
This function returns the attributes of a list of devices.

Change-Id: I322fc7db9c71e7c21fd03d616937d172da856428
Related-Bug: #1804274
changes/12/629912/7
Rodolfo Alonso Hernandez 4 years ago
parent 56dfd7291d
commit e7a2b6d179
3 changed files with 274 additions and 14 deletions
Show Stats Download Patch File Download Diff File

52
neutron/agent/linux/ip_lib.py
Unescape View File

@ -144,6 +144,19 @@ class IPWrapper(SubProcessBase):
def device(self, name):
return IPDevice(name, namespace=self.namespace)
def get_devices_info(self, exclude_loopback=True,
exclude_fb_tun_devices=True):
devices = get_devices_info(self.namespace)
retval = []
for device in devices:
if (exclude_loopback and device['name'] == LOOPBACK_DEVNAME or
exclude_fb_tun_devices and
device['name'] in FB_TUNNEL_DEVICE_NAMES):
continue
retval.append(device)
return retval
def get_devices(self, exclude_loopback=True, exclude_fb_tun_devices=True):
retval = []
try:
@ -1330,6 +1343,13 @@ def delete_ip_rule(namespace, ip, iif=None, table=None, priority=None,
privileged.delete_ip_rule(namespace, **cmd_args)
def get_attr(pyroute2_obj, attr_name):
"""Get an attribute from a PyRoute2 object"""
rule_attrs = pyroute2_obj.get('attrs', [])
for attr in (attr for attr in rule_attrs if attr[0] == attr_name):
return attr[1]
def _parse_link_device(namespace, device, **kwargs):
"""Parse pytoute2 link device information
@ -1337,12 +1357,6 @@ def _parse_link_device(namespace, device, **kwargs):
in a dictionary.
IP address scope: http://linux-ip.net/html/tools-ip-address.html
"""
def get_attr(pyroute2_obj, attr_name):
rule_attrs = pyroute2_obj.get('attrs', [])
for attr in (attr for attr in rule_attrs if attr[0] == attr_name):
return attr[1]
return
retval = []
name = get_attr(device, 'IFLA_IFNAME')
ip_addresses = privileged.get_ip_addresses(namespace,
@ -1377,3 +1391,29 @@ def get_devices_with_ip(namespace, name=None, **kwargs):
for device in devices):
retval += parsed_ips
return retval
def get_devices_info(namespace, **kwargs):
devices = privileged.get_link_devices(namespace, **kwargs)
retval = []
for device in devices:
ret = {'index': device['index'],
'name': get_attr(device, 'IFLA_IFNAME'),
'operstate': get_attr(device, 'IFLA_OPERSTATE'),
'linkmode': get_attr(device, 'IFLA_LINKMODE'),
'mtu': get_attr(device, 'IFLA_MTU'),
'promiscuity': get_attr(device, 'IFLA_PROMISCUITY'),
'mac': get_attr(device, 'IFLA_ADDRESS'),
'broadcast': get_attr(device, 'IFLA_BROADCAST')}
ifla_linkinfo = get_attr(device, 'IFLA_LINKINFO')
if ifla_linkinfo:
ret['kind'] = get_attr(ifla_linkinfo, 'IFLA_INFO_KIND')
ifla_data = get_attr(ifla_linkinfo, 'IFLA_INFO_DATA')
if ret['kind'] == 'vxlan':
ret['vxlan_id'] = get_attr(ifla_data, 'IFLA_VXLAN_ID')
ret['vxlan_group'] = get_attr(ifla_data, 'IFLA_VXLAN_GROUP')
elif ret['kind'] == 'vlan':
ret['vlan_id'] = get_attr(ifla_data, 'IFLA_VLAN_ID')
retval.append(ret)
return retval

124
neutron/tests/functional/privileged/agent/linux/test_ip_lib.py
Unescape View File

@ -21,13 +21,6 @@ from neutron.privileged.agent.linux import ip_lib as priv_ip_lib
from neutron.tests.functional import base as functional_base
def _get_attr(pyroute2_obj, attr_name):
rule_attrs = pyroute2_obj.get('attrs', [])
for attr in (attr for attr in rule_attrs if attr[0] == attr_name):
return attr[1]
return
class GetDeviceNamesTestCase(functional_base.BaseSudoTestCase):
def _remove_ns(self, namespace):
@ -57,6 +50,121 @@ class GetDeviceNamesTestCase(functional_base.BaseSudoTestCase):
self.assertNotIn(name, interfaces)
class GetDevicesInfoTestCase(functional_base.BaseSudoTestCase):
def setUp(self):
super(GetDevicesInfoTestCase, self).setUp()
self.namespace = 'ns_test-' + uuidutils.generate_uuid()
priv_ip_lib.create_netns(self.namespace)
self.addCleanup(self._remove_ns, self.namespace)
self.interfaces = ['int_01', 'int_02']
self.interfaces_to_exclude = (ip_lib.FB_TUNNEL_DEVICE_NAMES +
[ip_lib.LOOPBACK_DEVNAME])
def _remove_ns(self, namespace):
priv_ip_lib.remove_netns(namespace)
def test_get_devices_info_lo(self):
devices = priv_ip_lib.get_link_devices(self.namespace)
self.assertGreater(len(devices), 0)
for device in devices:
if ip_lib.get_attr(device, 'IFLA_IFNAME') != 'lo':
continue
self.assertIsNone(ip_lib.get_attr(device, 'IFLA_LINKINFO'))
break
else:
self.fail('Device "lo" not found')
def test_get_devices_info_dummy(self):
interfaces_tested = []
for interface in self.interfaces:
priv_ip_lib.create_interface(interface, self.namespace, 'dummy')
devices = priv_ip_lib.get_link_devices(self.namespace)
self.assertGreater(len(devices), 0)
for device in devices:
name = ip_lib.get_attr(device, 'IFLA_IFNAME')
if name in self.interfaces_to_exclude:
continue
self.assertIn(name, self.interfaces)
ifla_linkinfo = ip_lib.get_attr(device, 'IFLA_LINKINFO')
self.assertEqual(ip_lib.get_attr(ifla_linkinfo, 'IFLA_INFO_KIND'),
'dummy')
interfaces_tested.append(name)
self.assertEqual(sorted(interfaces_tested), sorted(self.interfaces))
def test_get_devices_info_vlan(self):
interfaces_tested = []
vlan_interfaces = []
vlan_id = 1000
for interface in self.interfaces:
priv_ip_lib.create_interface(interface, self.namespace, 'dummy')
vlan_interface = interface + '_' + str(vlan_id)
vlan_interfaces.append(vlan_interface)
priv_ip_lib.create_interface(
vlan_interface, self.namespace, 'vlan',
physical_interface=interface, vlan_id=vlan_id)
vlan_id += 1
devices = priv_ip_lib.get_link_devices(self.namespace)
self.assertGreater(len(devices), 0)
for device in devices:
name = ip_lib.get_attr(device, 'IFLA_IFNAME')
if name in self.interfaces_to_exclude:
continue
self.assertIn(name, self.interfaces + vlan_interfaces)
ifla_linkinfo = ip_lib.get_attr(device, 'IFLA_LINKINFO')
if name in vlan_interfaces:
self.assertEqual(
ip_lib.get_attr(ifla_linkinfo, 'IFLA_INFO_KIND'), 'vlan')
ifla_infodata = ip_lib.get_attr(ifla_linkinfo,
'IFLA_INFO_DATA')
vlan_id = int(name.split('_')[-1])
self.assertEqual(
ip_lib.get_attr(ifla_infodata, 'IFLA_VLAN_ID'), vlan_id)
interfaces_tested.append(name)
self.assertEqual(sorted(interfaces_tested),
sorted(self.interfaces + vlan_interfaces))
def test_get_devices_info_vxlan(self):
interfaces_tested = []
vxlan_interfaces = []
vxlan_id = 1000
for interface in self.interfaces:
priv_ip_lib.create_interface(interface, self.namespace, 'dummy')
vxlan_interface = interface + '_' + str(vxlan_id)
vxlan_interfaces.append(vxlan_interface)
priv_ip_lib.create_interface(
vxlan_interface, self.namespace, 'vxlan',
physical_interface=interface, vxlan_id=vxlan_id,
vxlan_group='239.1.1.1')
vxlan_id += 1
devices = priv_ip_lib.get_link_devices(self.namespace)
self.assertGreater(len(devices), 0)
for device in devices:
name = ip_lib.get_attr(device, 'IFLA_IFNAME')
if name in self.interfaces_to_exclude:
continue
self.assertIn(name, self.interfaces + vxlan_interfaces)
ifla_linkinfo = ip_lib.get_attr(device, 'IFLA_LINKINFO')
if name in vxlan_interfaces:
self.assertEqual(
ip_lib.get_attr(ifla_linkinfo, 'IFLA_INFO_KIND'),
'vxlan')
ifla_infodata = ip_lib.get_attr(ifla_linkinfo,
'IFLA_INFO_DATA')
vxlan_id = int(name.split('_')[-1])
self.assertEqual(
ip_lib.get_attr(ifla_infodata, 'IFLA_VXLAN_ID'), vxlan_id)
self.assertEqual(
ip_lib.get_attr(ifla_infodata, 'IFLA_VXLAN_GROUP'),
'239.1.1.1')
interfaces_tested.append(name)
self.assertEqual(sorted(interfaces_tested),
sorted(self.interfaces + vxlan_interfaces))
class ListIpRulesTestCase(functional_base.BaseSudoTestCase):
RULE_TABLES = {'default': 253, 'main': 254, 'local': 255}
@ -276,7 +384,7 @@ class GetIpAddressesTestCase(functional_base.BaseSudoTestCase):
ip_addresses = priv_ip_lib.get_ip_addresses(namespace)
for ip_address in ip_addresses:
int_name = str(ip_address['index'])
ip = _get_attr(ip_address, 'IFA_ADDRESS')
ip = ip_lib.get_attr(ip_address, 'IFA_ADDRESS')
mask = ip_address['prefixlen']
cidr = common_utils.ip_to_cidr(ip, mask)
self.assertEqual(interfaces[int_name]['cidr'], cidr)

112
neutron/tests/unit/agent/linux/test_ip_lib.py
Unescape View File

@ -1906,3 +1906,115 @@ class ParseLinkDeviceTestCase(base.BaseTestCase):
'dynamic': False, 'dadfailed': False, 'name': 'int_name',
'broadcast': None, 'tentative': False}]
self.assertEqual(expected, retval)
class GetDevicesInfoTestCase(base.BaseTestCase):
DEVICE_LO = {
'index': 2,
'attrs': (('IFLA_IFNAME', 'lo'), ('IFLA_OPERSTATE', 'UP'),
('IFLA_LINKMODE', 0), ('IFLA_MTU', 1000),
('IFLA_PROMISCUITY', 0),
('IFLA_ADDRESS', '5a:76:ed:cc:ce:90'),
('IFLA_BROADCAST', 'ff:ff:ff:ff:ff:f0'), )
}
DEVICE_DUMMY = {
'index': 2,
'attrs': (('IFLA_IFNAME', 'int_01'), ('IFLA_OPERSTATE', 'DOWN'),
('IFLA_LINKMODE', 0), ('IFLA_MTU', 1500),
('IFLA_PROMISCUITY', 0),
('IFLA_ADDRESS', '5a:76:ed:cc:ce:90'),
('IFLA_BROADCAST', 'ff:ff:ff:ff:ff:f0'),
('IFLA_LINKINFO', {
'attrs': (('IFLA_INFO_KIND', 'dummy'), )}))
}
DEVICE_VLAN = {
'index': 5,
'attrs': (('IFLA_IFNAME', 'int_02'), ('IFLA_OPERSTATE', 'DOWN'),
('IFLA_LINKMODE', 0), ('IFLA_MTU', 1400),
('IFLA_PROMISCUITY', 0),
('IFLA_ADDRESS', '5a:76:ed:cc:ce:91'),
('IFLA_BROADCAST', 'ff:ff:ff:ff:ff:f1'),
('IFLA_LINKINFO', {'attrs': (
('IFLA_INFO_KIND', 'vlan'),
('IFLA_INFO_DATA', {'attrs': (('IFLA_VLAN_ID', 1000), )})
)}))
}
DEVICE_VXLAN = {
'index': 9,
'attrs': (('IFLA_IFNAME', 'int_03'), ('IFLA_OPERSTATE', 'UP'),
('IFLA_LINKMODE', 0), ('IFLA_MTU', 1300),
('IFLA_PROMISCUITY', 0),
('IFLA_ADDRESS', '5a:76:ed:cc:ce:92'),
('IFLA_BROADCAST', 'ff:ff:ff:ff:ff:f2'),
('IFLA_LINKINFO', {'attrs': (
('IFLA_INFO_KIND', 'vxlan'),
('IFLA_INFO_DATA', {'attrs': (
('IFLA_VXLAN_ID', 1001),
('IFLA_VXLAN_GROUP', '239.1.1.1'))})
)}))
}
def setUp(self):
super(GetDevicesInfoTestCase, self).setUp()
self.mock_getdevs = mock.patch.object(priv_lib,
'get_link_devices').start()
def test_get_devices_info_lo(self):
self.mock_getdevs.return_value = (self.DEVICE_LO, )
ret = ip_lib.get_devices_info('namespace')
expected = {'index': 2,
'name': 'lo',
'operstate': 'UP',
'linkmode': 0,
'mtu': 1000,
'promiscuity': 0,
'mac': '5a:76:ed:cc:ce:90',
'broadcast': 'ff:ff:ff:ff:ff:f0'}
self.assertEqual(expected, ret[0])
def test_get_devices_info_dummy(self):
self.mock_getdevs.return_value = (self.DEVICE_DUMMY, )
ret = ip_lib.get_devices_info('namespace')
expected = {'index': 2,
'name': 'int_01',
'operstate': 'DOWN',
'linkmode': 0,
'mtu': 1500,
'promiscuity': 0,
'mac': '5a:76:ed:cc:ce:90',
'broadcast': 'ff:ff:ff:ff:ff:f0',
'kind': 'dummy'}
self.assertEqual(expected, ret[0])
def test_get_devices_info_vlan(self):
self.mock_getdevs.return_value = (self.DEVICE_VLAN, )
ret = ip_lib.get_devices_info('namespace')
expected = {'index': 5,
'name': 'int_02',
'operstate': 'DOWN',
'linkmode': 0,
'mtu': 1400,
'promiscuity': 0,
'mac': '5a:76:ed:cc:ce:91',
'broadcast': 'ff:ff:ff:ff:ff:f1',
'kind': 'vlan',
'vlan_id': 1000}
self.assertEqual(expected, ret[0])
def test_get_devices_info_vxlan(self):
self.mock_getdevs.return_value = (self.DEVICE_VXLAN, )
ret = ip_lib.get_devices_info('namespace')
expected = {'index': 9,
'name': 'int_03',
'operstate': 'UP',
'linkmode': 0,
'mtu': 1300,
'promiscuity': 0,
'mac': '5a:76:ed:cc:ce:92',
'broadcast': 'ff:ff:ff:ff:ff:f2',
'kind': 'vxlan',
'vxlan_id': 1001,
'vxlan_group': '239.1.1.1'}
self.assertEqual(expected, ret[0])

Write Preview
Loading…
Cancel
Save