openstack
/
neutron-vpnaas
Code Issues Proposed changes

Merge "Fix VPN Service for Distributed Routers"

This commit is contained in:
Jenkins 2015-01-20 17:58:44 +00:00 committed by Gerrit Code Review
parent 88e2d801d5 bd12e68b64
commit 08b9806d27
3 changed files with 78 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
neutron_vpnaas/db/vpn/vpn_db.py
View File

@ -619,7 +619,13 @@ class VPNPluginRpcDbMixin(object):
plugin = manager.NeutronManager.get_plugin()
agent = plugin._get_agent_by_type_and_host(
context, n_constants.AGENT_TYPE_L3, host)
if not agent.admin_state_up:
agent_conf = plugin.get_configuration_dict(agent)
# Retreive the agent_mode to check if this is the
# right agent to deploy the vpn service. In the
# case of distributed the vpn service should reside
# only on a dvr_snat node.
agent_mode = agent_conf.get('agent_mode', 'legacy')
if not agent.admin_state_up or agent_mode == 'dvr':
return []
query = context.session.query(VPNService)
query = query.join(IPsecSiteConnection)

36
neutron_vpnaas/services/vpn/vpn_service.py
View File

@ -84,7 +84,30 @@ class VPNService(advanced_service.AdvancedService):
router_info = self.l3_agent.router_info.get(router_id)
if not router_info:
return
return router_info.ns_name
# Added for handling the distributed Routers within SNAT namespace
if router_info.router['distributed']:
#return self.get_snat_ns_name(router_id)
return self.l3_agent.get_snat_ns_name(router_id)
else:
return router_info.ns_name
def get_router_based_iptables_manager(self, router_info):
"""Returns router based iptables manager
In DVR routers the IPsec VPN service should run inside
the snat namespace. So the iptables manager used for
snat namespace is different from the iptables manager
used for the qr namespace in a non dvr based router.
This function will check the router type and then will
return the right iptables manager. If DVR enabled router
it will return the snat_iptables_manager otherwise it will
return the legacy iptables_manager.
"""
if router_info.router['distributed']:
return router_info.snat_iptables_manager
else:
return router_info.iptables_manager
def add_nat_rule(self, router_id, chain, rule, top=False):
"""Add nat rule in namespace.
@ -99,8 +122,8 @@ class VPNService(advanced_service.AdvancedService):
router_info = self.l3_agent.router_info.get(router_id)
if not router_info:
return
router_info.iptables_manager.ipv4['nat'].add_rule(
chain, rule, top=top)
iptables_manager = self.get_router_based_iptables_manager(router_info)
iptables_manager.ipv4['nat'].add_rule(chain, rule, top=top)
def remove_nat_rule(self, router_id, chain, rule, top=False):
"""Remove nat rule in namespace.
@ -114,8 +137,8 @@ class VPNService(advanced_service.AdvancedService):
router_info = self.l3_agent.router_info.get(router_id)
if not router_info:
return
router_info.iptables_manager.ipv4['nat'].remove_rule(
chain, rule, top=top)
iptables_manager = self.get_router_based_iptables_manager(router_info)
iptables_manager.ipv4['nat'].remove_rule(chain, rule, top=top)
def iptables_apply(self, router_id):
"""Apply IPtables.
@ -126,4 +149,5 @@ class VPNService(advanced_service.AdvancedService):
router_info = self.l3_agent.router_info.get(router_id)
if not router_info:
return
router_info.iptables_manager.apply()
iptables_manager = self.get_router_based_iptables_manager(router_info)
iptables_manager.apply()

41
neutron_vpnaas/tests/unit/services/vpn/test_vpn_service.py
View File

@ -18,6 +18,7 @@ from oslo.config import cfg
from neutron.agent.common import config as agent_config
from neutron.agent.l3 import router_info
from neutron.agent.linux import iptables_manager
from neutron.extensions import vpnaas
from neutron.openstack.common import uuidutils
from neutron_vpnaas.services.vpn import agent as vpn_agent
@ -93,17 +94,38 @@ class TestVPNDeviceDriverCallsToService(base.BaseTestCase):
def _make_router_info_for_test(self, ns_name=None, iptables=None):
ri = router_info.RouterInfo(FAKE_ROUTER_ID, self.conf.root_helper,
{}, ns_name=ns_name)
ri.router['distributed'] = False
if iptables:
ri.iptables_manager.ipv4['nat'] = iptables
ri.iptables_manager.apply = self.apply_mock
self.service.l3_agent.router_info = {FAKE_ROUTER_ID: ri}
def _make_dvr_router_info_for_test(self, ns_name=None, iptables=None):
ri = router_info.RouterInfo(FAKE_ROUTER_ID, self.conf.root_helper,
{}, ns_name=ns_name)
ri.router['distributed'] = True
if iptables:
ri.snat_iptables_manager = iptables_manager.IptablesManager(
root_helper=mock.ANY,
namespace='snat-' + FAKE_ROUTER_ID,
use_ipv6=mock.ANY)
ri.snat_iptables_manager.ipv4['nat'] = iptables
ri.snat_iptables_manager.apply = self.apply_mock
self.service.l3_agent.router_info = {FAKE_ROUTER_ID: ri}
def test_get_namespace_for_router(self):
ns = "ns-" + FAKE_ROUTER_ID
self._make_router_info_for_test(ns_name=ns)
namespace = self.service.get_namespace(FAKE_ROUTER_ID)
self.assertTrue(namespace.endswith(FAKE_ROUTER_ID))
def test_get_namespace_for_dvr_router(self):
ns = "ns-" + FAKE_ROUTER_ID
self._make_dvr_router_info_for_test(ns_name=ns)
namespace = self.service.get_namespace(FAKE_ROUTER_ID)
self.assertTrue(namespace.startswith('snat'))
self.assertTrue(namespace.endswith(FAKE_ROUTER_ID))
def test_fail_getting_namespace_for_unknown_router(self):
self._make_router_info_for_test()
self.assertFalse(self.service.get_namespace('bogus_id'))
@ -115,6 +137,13 @@ class TestVPNDeviceDriverCallsToService(base.BaseTestCase):
self.iptables.add_rule.assert_called_once_with(
'fake_chain', 'fake_rule', top=True)
def test_add_nat_rule_with_dvr_router(self):
self._make_dvr_router_info_for_test(iptables=self.iptables)
self.service.add_nat_rule(FAKE_ROUTER_ID, 'fake_chain',
'fake_rule', True)
self.iptables.add_rule.assert_called_once_with(
'fake_chain', 'fake_rule', top=True)
def test_add_nat_rule_with_no_router(self):
self._make_router_info_for_test(iptables=self.iptables)
self.service.add_nat_rule(
@ -131,6 +160,13 @@ class TestVPNDeviceDriverCallsToService(base.BaseTestCase):
self.iptables.remove_rule.assert_called_once_with(
'fake_chain', 'fake_rule', top=True)
def test_remove_rule_with_dvr_router(self):
self._make_router_info_for_test(iptables=self.iptables)
self.service.remove_nat_rule(FAKE_ROUTER_ID, 'fake_chain',
'fake_rule', True)
self.iptables.remove_rule.assert_called_once_with(
'fake_chain', 'fake_rule', top=True)
def test_remove_rule_with_no_router(self):
self._make_router_info_for_test(iptables=self.iptables)
self.service.remove_nat_rule(
@ -144,6 +180,11 @@ class TestVPNDeviceDriverCallsToService(base.BaseTestCase):
self.service.iptables_apply(FAKE_ROUTER_ID)
self.apply_mock.assert_called_once_with()
def test_iptables_apply_with_dvr_router(self):
self._make_router_info_for_test(iptables=self.iptables)
self.service.iptables_apply(FAKE_ROUTER_ID)
self.apply_mock.assert_called_once_with()
def test_iptables_apply_with_no_router(self):
self._make_router_info_for_test(iptables=self.iptables)
self.service.iptables_apply('bogus_router_id')