import eventlet
eventlet.monkey_patch()
-import netaddr
from oslo.config import cfg
from oslo import messaging
from oslo.utils import importutils
# or all the networks connected via a router
# to the one passed as a parameter
neutron_lookup_param = '--network_id=%s' % network.id
- meta_cidr = netaddr.IPNetwork(dhcp.METADATA_DEFAULT_CIDR)
- has_metadata_subnet = any(netaddr.IPNetwork(s.cidr) in meta_cidr
- for s in network.subnets)
- if (self.conf.enable_metadata_network and has_metadata_subnet):
+ # When the metadata network is enabled, the proxy might
+ # be started for the router attached to the network
+ if self.conf.enable_metadata_network:
router_ports = [port for port in network.ports
if (port.device_owner ==
constants.DEVICE_OWNER_ROUTER_INTF)]
@classmethod
def should_enable_metadata(cls, conf, network):
- """True if there exists a subnet for which a metadata proxy is needed
+ """Determine whether the metadata proxy is needed for a network
+
+ This method returns True for truly isolated networks (ie: not attached
+ to a router), when the enable_isolated_metadata flag is True.
+
+ This method also returns True when enable_metadata_network is True,
+ and the network passed as a parameter has a subnet in the link-local
+ CIDR, thus characterizing it as a "metadata" network. The metadata
+ network is used by solutions which do not leverage the l3 agent for
+ providing access to the metadata service via logical routers built
+ with 3rd party backends.
"""
+ if conf.enable_metadata_network and conf.enable_isolated_metadata:
+ # check if the network has a metadata subnet
+ meta_cidr = netaddr.IPNetwork(METADATA_DEFAULT_CIDR)
+ if any(netaddr.IPNetwork(s.cidr) in meta_cidr
+ for s in network.subnets):
+ return True
+
if not conf.use_namespaces or not conf.enable_isolated_metadata:
return False
id = 'rrrrrrrr-rrrr-rrrr-rrrr-rrrrrrrrrrrr'
admin_state_up = True
device_owner = constants.DEVICE_OWNER_ROUTER_INTF
- fixed_ips = [FakeIPAllocation('192.168.0.1',
- 'dddddddd-dddd-dddd-dddd-dddddddddddd')]
mac_address = '00:00:0f:rr:rr:rr'
- def __init__(self, dev_owner=constants.DEVICE_OWNER_ROUTER_INTF):
+ def __init__(self, dev_owner=constants.DEVICE_OWNER_ROUTER_INTF,
+ ip_address='192.168.0.1'):
self.extra_dhcp_opts = []
self.device_owner = dev_owner
+ self.fixed_ips = [FakeIPAllocation(
+ ip_address, 'dddddddd-dddd-dddd-dddd-dddddddddddd')]
class FakePortMultipleAgents1:
dns_nameservers = ['8.8.8.8']
+class FakeV4MetadataSubnet:
+ id = 'dddddddd-dddd-dddd-dddd-dddddddddddd'
+ ip_version = 4
+ cidr = '169.254.169.254/30'
+ gateway_ip = '169.254.169.253'
+ enable_dhcp = True
+ host_routes = []
+ dns_nameservers = []
+
+
class FakeV4SubnetGatewayRoute:
id = 'dddddddd-dddd-dddd-dddd-dddddddddddd'
ip_version = 4
ports = [FakePort1()]
+class FakeV4MetadataNetwork:
+ id = 'cccccccc-cccc-cccc-cccc-cccccccccccc'
+ subnets = [FakeV4MetadataSubnet()]
+ ports = [FakeRouterPort(ip_address='169.254.169.253')]
+
+
class FakeV4NetworkDistRouter:
id = 'cccccccc-cccc-cccc-cccc-cccccccccccc'
subnets = [FakeV4Subnet()]
self.conf.register_opts(base_config.core_opts)
self.conf.register_opts(dhcp.OPTS)
config.register_interface_driver_opts_helper(self.conf)
+ config.register_use_namespaces_opts_helper(self.conf)
instance = mock.patch("neutron.agent.linux.dhcp.DeviceManager")
self.mock_mgr = instance.start()
self.conf.register_opt(cfg.BoolOpt('enable_isolated_metadata',
default=True))
+ self.conf.register_opt(cfg.BoolOpt('enable_metadata_network',
+ default=False))
self.config_parse(self.conf)
self.conf.set_override('state_path', '')
- self.conf.use_namespaces = True
self.replace_p = mock.patch('neutron.agent.linux.utils.replace_file')
self.execute_p = mock.patch('neutron.agent.linux.utils.execute')
dm._output_hosts_file()
self.safe.assert_has_calls([mock.call(exp_host_name,
exp_host_data)])
+
+ def test_should_enable_metadata_namespaces_disabled_returns_false(self):
+ self.conf.set_override('use_namespaces', False)
+ self.assertFalse(dhcp.Dnsmasq.should_enable_metadata(self.conf,
+ mock.ANY))
+
+ def test_should_enable_metadata_isolated_network_returns_true(self):
+ self.assertTrue(dhcp.Dnsmasq.should_enable_metadata(
+ self.conf, FakeV4NetworkNoRouter()))
+
+ def test_should_enable_metadata_non_isolated_network_returns_false(self):
+ self.assertFalse(dhcp.Dnsmasq.should_enable_metadata(
+ self.conf, FakeV4NetworkDistRouter()))
+
+ def test_should_enable_metadata_isolated_meta_disabled_returns_false(self):
+ self.conf.set_override('enable_isolated_metadata', False)
+ self.assertFalse(dhcp.Dnsmasq.should_enable_metadata(self.conf,
+ mock.ANY))
+
+ def test_should_enable_metadata_with_metadata_network_returns_true(self):
+ self.conf.set_override('enable_metadata_network', True)
+ self.assertTrue(dhcp.Dnsmasq.should_enable_metadata(
+ self.conf, FakeV4MetadataNetwork()))