class SecurityGroupAgentRpc(object):
"""Enables SecurityGroup agent support in agent implementations."""
- def __init__(self, context, plugin_rpc, root_helper,
- defer_refresh_firewall=False):
+ def __init__(self, context, plugin_rpc, defer_refresh_firewall=False):
self.context = context
self.plugin_rpc = plugin_rpc
- self.root_helper = root_helper
self.init_firewall(defer_refresh_firewall)
def init_firewall(self, defer_refresh_firewall=False):
target = oslo_messaging.Target(version='1.1')
- def __init__(self, integ_br, polling_interval, root_helper, vs='ovs'):
+ def __init__(self, integ_br, polling_interval, vs='ovs'):
super(RestProxyAgent, self).__init__()
self.polling_interval = polling_interval
self._setup_rpc()
self.sg_agent = sg_rpc.SecurityGroupAgentRpc(self.context,
- self.sg_plugin_rpc,
- root_helper)
+ self.sg_plugin_rpc)
if vs == 'ivs':
self.int_br = IVSBridge(integ_br)
else:
integ_br = cfg.CONF.RESTPROXYAGENT.integration_bridge
polling_interval = cfg.CONF.RESTPROXYAGENT.polling_interval
- root_helper = cfg.CONF.AGENT.root_helper
- bsnagent = RestProxyAgent(integ_br, polling_interval, root_helper,
+ bsnagent = RestProxyAgent(integ_br, polling_interval,
cfg.CONF.RESTPROXYAGENT.virtual_switch_type)
bsnagent.daemon_loop()
sys.exit(0)
class HyperVSecurityAgent(sg_rpc.SecurityGroupAgentRpc):
def __init__(self, context, plugin_rpc):
- # Note: as rootwrap is not supported on HyperV, root_helper is
- # passed in as None.
- super(HyperVSecurityAgent, self).__init__(context, plugin_rpc,
- root_helper=None)
+ super(HyperVSecurityAgent, self).__init__(context, plugin_rpc)
if sg_rpc.is_firewall_enabled():
self._setup_rpc()
class NECNeutronAgent(object):
- def __init__(self, integ_br, root_helper, polling_interval):
+ def __init__(self, integ_br, polling_interval):
'''Constructor.
:param integ_br: name of the integration bridge.
- :param root_helper: utility to use when running shell cmds.
:param polling_interval: interval (secs) to check the bridge.
'''
self.int_br = ovs_lib.OVSBridge(integ_br)
'agent_type': q_const.AGENT_TYPE_NEC,
'start_flag': True}
- self.setup_rpc(root_helper)
+ self.setup_rpc()
- def setup_rpc(self, root_helper):
+ def setup_rpc(self):
self.host = socket.gethostname()
self.agent_id = 'nec-q-agent.%s' % self.host
LOG.info(_LI("RPC agent_id: %s"), self.agent_id)
self.state_rpc = agent_rpc.PluginReportStateAPI(topics.PLUGIN)
self.sg_plugin_rpc = sg_rpc.SecurityGroupServerRpcApi(topics.PLUGIN)
self.sg_agent = sg_rpc.SecurityGroupAgentRpc(self.context,
- self.sg_plugin_rpc, root_helper)
+ self.sg_plugin_rpc)
# RPC network init
# Handle updates from service
# Determine which agent type to use.
integ_br = config.OVS.integration_bridge
- root_helper = config.AGENT.root_helper
polling_interval = config.AGENT.polling_interval
- agent = NECNeutronAgent(integ_br, root_helper, polling_interval)
+ agent = NECNeutronAgent(integ_br, polling_interval)
# Start everything.
agent.daemon_loop()
target = oslo_messaging.Target(version='1.1')
def __init__(self, ryuapp, integ_br, local_ip,
- bridge_mappings, interface_mappings, root_helper,
+ bridge_mappings, interface_mappings,
polling_interval, tunnel_types=None):
"""Constructor.
(deprecated)
:param interface_mappings: mappings from physical network name to
interface.
- :param root_helper: utility to use when running shell cmds.
:param polling_interval: interval (secs) to poll DB.
:param tunnel_types: A list of tunnel types to enable support for in
the agent. If set, will automatically set enable_tunneling to
"""
super(OFANeutronAgent, self).__init__()
self.ryuapp = ryuapp
- self.root_helper = root_helper
# TODO(yamamoto): Remove this VLAN leftover
self.available_local_vlans = set(xrange(ofa_const.LOCAL_VLAN_MIN,
ofa_const.LOCAL_VLAN_MAX))
# Security group agent support
self.sg_agent = sg_rpc.SecurityGroupAgentRpc(self.context,
- self.sg_plugin_rpc, self.root_helper,
- defer_refresh_firewall=True)
+ self.sg_plugin_rpc, defer_refresh_firewall=True)
# Initialize iteration counter
self.iter_num = 0
local_ip=config.OVS.local_ip,
interface_mappings=interface_mappings,
bridge_mappings=bridge_mappings,
- root_helper=config.AGENT.root_helper,
polling_interval=config.AGENT.polling_interval,
tunnel_types=config.AGENT.tunnel_types,
)
# 1.1 Support Security Group RPC
target = oslo_messaging.Target(version='1.1')
- def __init__(self, integ_br, root_helper, polling_interval):
+ def __init__(self, integ_br, polling_interval):
super(NVSDNeutronAgent, self).__init__()
self.int_br = ovs_lib.OVSBridge(integ_br)
self.polling_interval = polling_interval
- self.root_helper = root_helper
self.setup_rpc()
self.ports = set()
self.context = n_context.get_admin_context_without_session()
self.sg_plugin_rpc = sg_rpc.SecurityGroupServerRpcApi(topics.PLUGIN)
self.sg_agent = sg_rpc.SecurityGroupAgentRpc(self.context,
- self.sg_plugin_rpc,
- self.root_helper)
+ self.sg_plugin_rpc)
# RPC network init
# Handle updates from service
common_config.setup_logging()
integ_br = config.AGENT.integration_bridge
- root_helper = config.AGENT.root_helper
polling_interval = config.AGENT.polling_interval
- agent = NVSDNeutronAgent(integ_br, root_helper, polling_interval)
+ agent = NVSDNeutronAgent(integ_br, polling_interval)
LOG.info(_LI("NVSD Agent initialized successfully, now running... "))
# Start everything.
# Security group agent support
self.sg_agent = sg_rpc.SecurityGroupAgentRpc(self.context,
- self.sg_plugin_rpc, root_helper, defer_refresh_firewall=True)
+ self.sg_plugin_rpc, defer_refresh_firewall=True)
# Initialize iteration counter
self.iter_num = 0
self.phys_brs = {}
self.int_ofports = {}
self.phys_ofports = {}
- ip_wrapper = ip_lib.IPWrapper(self.root_helper)
+ ip_wrapper = ip_lib.IPWrapper()
ovs = ovs_lib.BaseOVS()
ovs_bridges = ovs.get_bridges()
for physical_network, bridge in bridge_mappings.iteritems():
self.int_br.delete_port(int_if_name)
br.delete_port(phys_if_name)
if self.use_veth_interconnection:
- if ip_lib.device_exists(int_if_name, self.root_helper):
- ip_lib.IPDevice(int_if_name,
- self.root_helper).link.delete()
+ if ip_lib.device_exists(int_if_name):
+ ip_lib.IPDevice(int_if_name).link.delete()
# Give udev a chance to process its rules here, to avoid
# race conditions between commands launched by udev rules
# and the subsequent call to ip_wrapper.add_veth
self.plugin_rpc = agent_rpc.PluginApi(topics.PLUGIN)
self.sg_plugin_rpc = sg_rpc.SecurityGroupServerRpcApi(topics.PLUGIN)
self.sg_agent = sg_rpc.SecurityGroupAgentRpc(self.context,
- self.sg_plugin_rpc, self.root_helper)
+ self.sg_plugin_rpc)
self._setup_rpc()
# Initialize iteration counter
self.iter_num = 0
def mock_agent(self):
mock_context = mock.Mock(return_value='abc')
self.context.get_admin_context_without_session = mock_context
- return self.mod_agent.RestProxyAgent('int-br', 2, 'helper')
+ return self.mod_agent.RestProxyAgent('int-br', 2)
def mock_port_update(self, **kwargs):
agent = self.mock_agent()
def mock_main(self):
cfg_attrs = {'CONF.RESTPROXYAGENT.integration_bridge': 'integ_br',
'CONF.RESTPROXYAGENT.polling_interval': 5,
- 'CONF.RESTPROXYAGENT.virtual_switch_type': 'ovs',
- 'CONF.AGENT.root_helper': 'helper'}
+ 'CONF.RESTPROXYAGENT.virtual_switch_type': 'ovs'}
with contextlib.nested(
mock.patch(AGENTMOD + '.cfg', **cfg_attrs),
mock.patch(AGENTMOD + '.config.init'),
self.assertRaises(SystemExit, self.mock_main)
mock_agent.assert_has_calls([
- mock.call('integ_br', 5, 'helper', 'ovs'),
+ mock.call('integ_br', 5, 'ovs'),
mock.call().daemon_loop()
])
mock.patch('neutron.agent.rpc.PluginReportStateAPI')
) as (get_datapath_id, gethostname,
loopingcall, state_rpc_api):
- kwargs = {'integ_br': 'integ_br',
- 'root_helper': 'dummy_wrapper',
- 'polling_interval': 1}
+ kwargs = {'integ_br': 'integ_br', 'polling_interval': 1}
self.agent = nec_neutron_agent.NECNeutronAgent(**kwargs)
self.loopingcall = loopingcall
self.state_rpc_api = state_rpc_api
mock.patch.object(nec_neutron_agent, 'config')
) as (agent, common_config, cfg):
cfg.OVS.integration_bridge = 'br-int-x'
- cfg.AGENT.root_helper = 'dummy-helper'
cfg.AGENT.polling_interval = 10
nec_neutron_agent.main()
self.assertTrue(common_config.setup_logging.called)
agent.assert_has_calls([
- mock.call('br-int-x', 'dummy-helper', 10),
+ mock.call('br-int-x', 10),
mock.call().daemon_loop()
])
'FixedIntervalLoopingCall'),
) as (loopingcall):
kwargs = {'integ_br': 'integration_bridge',
- 'root_helper': 'dummy_wrapper',
'polling_interval': 5}
context = mock.Mock()
self.agent = nvsd_neutron_agent.NVSDNeutronAgent(**kwargs)
sg_plugin_rpc = sg_rpc.SecurityGroupServerRpcApi(topics.PLUGIN)
self.sg_agent = sg_rpc.SecurityGroupAgentRpc(context,
- sg_plugin_rpc, 'dummy_wrapper')
+ sg_plugin_rpc)
self.callback_nvsd = nvsd_neutron_agent.NVSDAgentRpcCallback(
context, self.agent, self.sg_agent)
self.loopingcall = loopingcall
mock.patch.object(nvsd_neutron_agent, 'config')
) as (agent, common_config, config):
config.AGENT.integration_bridge = 'br-int-dummy'
- config.AGENT.root_helper = 'root-helper'
config.AGENT.polling_interval = 5
nvsd_neutron_agent.main()
self.assertTrue(common_config.setup_logging.called)
agent.assert_has_calls([
- mock.call('br-int-dummy', 'root-helper', 5),
+ mock.call('br-int-dummy', 5),
mock.call().daemon_loop()
])
self.device_exists_expected = []
self.ipdevice_expected = []
- self.ipwrapper_expected = [mock.call('sudo')]
+ self.ipwrapper_expected = [mock.call()]
self.get_bridges_expected = [mock.call(), mock.call()]
]
self.device_exists_expected = [
- mock.call('int-%s' % self.MAP_TUN_BRIDGE, 'sudo'),
+ mock.call('int-%s' % self.MAP_TUN_BRIDGE),
]
self.ipdevice_expected = [
- mock.call('int-%s' % self.MAP_TUN_BRIDGE, 'sudo'),
+ mock.call('int-%s' % self.MAP_TUN_BRIDGE),
mock.call().link.delete()
]
self.ipwrapper_expected = [
- mock.call('sudo'),
+ mock.call(),
mock.call().add_veth('int-%s' % self.MAP_TUN_BRIDGE,
'phy-%s' % self.MAP_TUN_BRIDGE)
]
def test_init_firewall_with_none_driver(self):
set_enable_security_groups(False)
agent = sg_rpc.SecurityGroupAgentRpc(
- context=None, plugin_rpc=mock.Mock(), root_helper=None)
+ context=None, plugin_rpc=mock.Mock())
self.assertEqual(agent.firewall.__class__.__name__,
'NoopFirewallDriver')
super(BaseSecurityGroupAgentRpcTestCase, self).setUp()
set_firewall_driver(FIREWALL_NOOP_DRIVER)
self.agent = sg_rpc.SecurityGroupAgentRpc(
- context=None, plugin_rpc=mock.Mock(), root_helper='sudo',
+ context=None, plugin_rpc=mock.Mock(),
defer_refresh_firewall=defer_refresh_firewall)
mock.patch('neutron.agent.linux.iptables_manager').start()
self.default_firewall = self.agent.firewall
def setUp(self, defer_refresh_firewall=False, test_rpc_v1_1=True):
super(TestSecurityGroupAgentWithIptables, self).setUp()
- config.register_root_helper(cfg.CONF)
config.register_iptables_opts(cfg.CONF)
set_firewall_driver(self.FIREWALL_DRIVER)
cfg.CONF.set_override('enable_ipset', False, group='SECURITYGROUP')
self.rpc = mock.Mock()
self.agent = sg_rpc.SecurityGroupAgentRpc(
- context=None, plugin_rpc=self.rpc, root_helper='sudo',
+ context=None, plugin_rpc=self.rpc,
defer_refresh_firewall=defer_refresh_firewall)
- self.root_helper = 'sudo'
if test_rpc_v1_1:
self.rpc.security_group_info_for_devices.side_effect = (
Code Review / openstack-build / neutron-build.git / commitdiff