openstack
/
neutron
Code Issues Proposed changes

Merge "DVR: Look at all SNAT ports for a subnet match"

This commit is contained in:
Jenkins 2017-02-10 04:09:13 +00:00 committed by Gerrit Code Review
parent 02f893a56d 0629129c03
commit cea029f6d3
3 changed files with 59 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
neutron/agent/l3/dvr_router_base.py
View File

@ -38,16 +38,17 @@ class DvrRouterBase(router.RouterInfo):
"""Return the SNAT port for the given internal interface port."""
if snat_ports is None:
snat_ports = self.get_snat_interfaces()
fixed_ip = int_port['fixed_ips'][0]
subnet_id = fixed_ip['subnet_id']
if snat_ports:
match_port = [p for p in snat_ports
if p['fixed_ips'][0]['subnet_id'] == subnet_id]
if match_port:
return match_port[0]
else:
LOG.error(_LE('DVR: SNAT port not found in the list '
'%(snat_list)s for the given router '
' internal port %(int_p)s'), {
'snat_list': snat_ports,
'int_p': int_port})
if not snat_ports:
return
fixed_ips = int_port['fixed_ips']
subnet_ids = [fixed_ip['subnet_id'] for fixed_ip in fixed_ips]
for p in snat_ports:
for ip in p['fixed_ips']:
if ip['subnet_id'] in subnet_ids:
return p
LOG.error(_LE('DVR: SNAT port not found in the list '
'%(snat_list)s for the given router '
'internal port %(int_p)s'), {
'snat_list': snat_ports,
'int_p': int_port})

15
neutron/tests/common/l3_test_common.py
View File

@ -132,18 +132,29 @@ def get_subnet_id(port):
def router_append_interface(router, count=1, ip_version=4, ra_mode=None,
addr_mode=None, dual_stack=False):
addr_mode=None, dual_stack=False, same_port=False):
interfaces = router[lib_constants.INTERFACE_KEY]
current = sum(
[netaddr.IPNetwork(subnet['cidr']).version == ip_version
for p in interfaces for subnet in p['subnets']])
# If dual_stack=True, create IPv4 and IPv6 subnets on each port
# If same_port=True, create ip_version number of subnets on a single port
# Else create just an ip_version subnet on each port
if dual_stack:
ip_versions = [4, 6]
elif same_port:
ip_versions = [ip_version] * count
count = 1
else:
ip_versions = [ip_version]
mac_address = netaddr.EUI('ca:fe:de:ad:be:ef')
mac_address.dialect = netaddr.mac_unix
for i in range(current, current + count):
fixed_ips = []
subnets = []
for loop_version in (4, 6):
for loop_version in ip_versions:
if loop_version == 4 and (ip_version == 4 or dual_stack):
ip_pool = '35.4.%i.4'
cidr_pool = '35.4.%i.0/24'

32
neutron/tests/unit/agent/l3/test_agent.py
View File

@ -887,6 +887,38 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
self.assertNotEqual(test_port, res_ip)
self.assertIsNone(res_ip)
def test_get_snat_port_for_internal_port_ipv6_same_port(self):
router = l3_test_common.prepare_router_data(ip_version=4,
enable_snat=True,
num_internal_ports=1)
ri = dvr_router.DvrEdgeRouter(mock.sentinel.agent,
HOSTNAME,
router['id'],
router,
**self.ri_kwargs)
# Add two additional IPv6 prefixes on the same interface
l3_test_common.router_append_interface(router, count=2, ip_version=6,
same_port=True)
internal_ports = ri.router.get(lib_constants.INTERFACE_KEY, [])
with mock.patch.object(ri, 'get_snat_interfaces') as get_interfaces:
get_interfaces.return_value = internal_ports
# get the second internal interface in the list
res_port = ri.get_snat_port_for_internal_port(internal_ports[1])
self.assertEqual(internal_ports[1], res_port)
# tweak the first subnet_id, should still find port based
# on second subnet_id
test_port = copy.deepcopy(res_port)
test_port['fixed_ips'][0]['subnet_id'] = 1234
res_ip = ri.get_snat_port_for_internal_port(test_port)
self.assertEqual(internal_ports[1], res_ip)
# tweak the second subnet_id, shouldn't match now
test_port['fixed_ips'][1]['subnet_id'] = 1234
res_ip = ri.get_snat_port_for_internal_port(test_port)
self.assertIsNone(res_ip)
def test_process_cent_router(self):
router = l3_test_common.prepare_router_data()
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)