# Example: physical_network_mtus = physnet1:1550, physnet2:1500
# ======== end of items for MTU selection and advertisement =========
+# (StrOpt) Default network type for external networks when no provider
+# attributes are specified. By default it is None, which means that if
+# provider attributes are not specified while creating external networks
+# then they will have the same type as tenant networks.
+# Allowed values for external_network_type config option depend on the
+# network type values configured in type_drivers config option.
+# external_network_type =
+# Example: external_network_type = local
+
[ml2_type_flat]
# (ListOpt) List of physical_network names with which flat networks
# can be created. Use * to allow flat networks with arbitrary
"<physnet>:<mtu val>. This mapping allows "
"specifying a physical network MTU value that "
"differs from the default segment_mtu value.")),
+ cfg.StrOpt('external_network_type',
+ help=_("Default network type for external networks when no "
+ "provider attributes are specified. By default it is "
+ "None, which means that if provider attributes are not "
+ "specified while creating external networks then they "
+ "will have the same type as tenant networks. Allowed "
+ "values for external_network_type config option depend "
+ "on the network type values configured in type_drivers "
+ "config option."))
]
from neutron.api.v2 import attributes
from neutron.common import exceptions as exc
+from neutron.extensions import external_net
from neutron.extensions import multiprovidernet as mpnet
from neutron.extensions import portbindings
from neutron.extensions import providernet as provider
LOG.info(_LI("Loaded type driver names: %s"), self.names())
self._register_types()
self._check_tenant_network_types(cfg.CONF.ml2.tenant_network_types)
+ self._check_external_network_type(cfg.CONF.ml2.external_network_type)
def _register_types(self):
for ext in self:
raise SystemExit(1)
LOG.info(_LI("Tenant network_types: %s"), self.tenant_network_types)
+ def _check_external_network_type(self, ext_network_type):
+ if ext_network_type and ext_network_type not in self.drivers:
+ LOG.error(_LE("No type driver for external network_type: %s. "
+ "Service terminated!"), ext_network_type)
+ raise SystemExit(1)
+
def _process_provider_segment(self, segment):
(network_type, physical_network,
segmentation_id) = (self._get_attribute(segment, attr)
elif attributes.is_attr_set(network.get(mpnet.SEGMENTS)):
segments = [self._process_provider_segment(s)
for s in network[mpnet.SEGMENTS]]
- mpnet.check_duplicate_segments(
- segments,
- self.is_partial_segment)
+ mpnet.check_duplicate_segments(segments, self.is_partial_segment)
return segments
def _match_segment(self, segment, filters):
LOG.info(_LI("Initializing driver for type '%s'"), network_type)
driver.obj.initialize()
+ def _add_network_segment(self, session, network_id, segment, mtu,
+ segment_index=0):
+ db.add_network_segment(session, network_id, segment, segment_index)
+ if segment.get(api.MTU) > 0:
+ mtu.append(segment[api.MTU])
+
def create_network_segments(self, context, network, tenant_id):
"""Call type drivers to create network segments."""
segments = self._process_provider_create(network)
for segment_index, segment in enumerate(segments):
segment = self.reserve_provider_segment(
session, segment)
- db.add_network_segment(session, network_id,
- segment, segment_index)
- if segment.get(api.MTU) > 0:
- mtu.append(segment[api.MTU])
+ self._add_network_segment(session, network_id, segment,
+ mtu, segment_index)
+ elif (cfg.CONF.ml2.external_network_type and
+ self._get_attribute(network, external_net.EXTERNAL)):
+ segment = self._allocate_ext_net_segment(session)
+ self._add_network_segment(session, network_id, segment, mtu)
else:
- segment = self.allocate_tenant_segment(session)
- db.add_network_segment(session, network_id, segment)
- if segment.get(api.MTU) > 0:
- mtu.append(segment[api.MTU])
+ segment = self._allocate_tenant_net_segment(session)
+ self._add_network_segment(session, network_id, segment, mtu)
network[api.MTU] = min(mtu) if mtu else 0
def is_partial_segment(self, segment):
driver = self.drivers.get(network_type)
return driver.obj.reserve_provider_segment(session, segment)
- def allocate_tenant_segment(self, session):
+ def _allocate_segment(self, session, network_type):
+ driver = self.drivers.get(network_type)
+ return driver.obj.allocate_tenant_segment(session)
+
+ def _allocate_tenant_net_segment(self, session):
for network_type in self.tenant_network_types:
- driver = self.drivers.get(network_type)
- segment = driver.obj.allocate_tenant_segment(session)
+ segment = self._allocate_segment(session, network_type)
if segment:
return segment
raise exc.NoNetworkAvailable()
+ def _allocate_ext_net_segment(self, session):
+ network_type = cfg.CONF.ml2.external_network_type
+ segment = self._allocate_segment(session, network_type)
+ if segment:
+ return segment
+ raise exc.NoNetworkAvailable()
+
def release_network_segments(self, session, network_id):
segments = db.get_network_segments(session, network_id,
filter_dynamic=None)
from neutron.db import db_base_plugin_v2 as base_plugin
from neutron.db import l3_db
from neutron.db import models_v2
-from neutron.extensions import external_net as external_net
+from neutron.extensions import external_net
from neutron.extensions import multiprovidernet as mpnet
from neutron.extensions import portbindings
from neutron.extensions import providernet as pnet
from neutron import manager
-from neutron.plugins.common import constants as service_constants
+from neutron.plugins.common import constants as p_const
from neutron.plugins.ml2.common import exceptions as ml2_exc
from neutron.plugins.ml2 import config
from neutron.plugins.ml2 import db as ml2_db
self.assertEqual(db_api.MAX_RETRIES + 1, f.call_count)
+class TestExternalNetwork(Ml2PluginV2TestCase):
+
+ def _create_external_network(self):
+ data = {'network': {'name': 'net1',
+ 'router:external': 'True',
+ 'tenant_id': 'tenant_one'}}
+ network_req = self.new_create_request('networks', data)
+ network = self.deserialize(self.fmt,
+ network_req.get_response(self.api))
+ return network
+
+ def test_external_network_type_none(self):
+ config.cfg.CONF.set_default('external_network_type',
+ None,
+ group='ml2')
+
+ network = self._create_external_network()
+ # For external network, expected network type to be
+ # tenant_network_types which is by default 'local'.
+ self.assertEqual(p_const.TYPE_LOCAL,
+ network['network'][pnet.NETWORK_TYPE])
+ # No physical network specified, expected 'None'.
+ self.assertIsNone(network['network'][pnet.PHYSICAL_NETWORK])
+ # External network will not have a segmentation id.
+ self.assertIsNone(network['network'][pnet.SEGMENTATION_ID])
+ # External network will not have multiple segments.
+ self.assertNotIn(mpnet.SEGMENTS, network['network'])
+
+ def test_external_network_type_vlan(self):
+ config.cfg.CONF.set_default('external_network_type',
+ p_const.TYPE_VLAN,
+ group='ml2')
+
+ network = self._create_external_network()
+ # For external network, expected network type to be 'vlan'.
+ self.assertEqual(p_const.TYPE_VLAN,
+ network['network'][pnet.NETWORK_TYPE])
+ # Physical network is expected.
+ self.assertIsNotNone(network['network'][pnet.PHYSICAL_NETWORK])
+ # External network will have a segmentation id.
+ self.assertIsNotNone(network['network'][pnet.SEGMENTATION_ID])
+ # External network will not have multiple segments.
+ self.assertNotIn(mpnet.SEGMENTS, network['network'])
+
+
class TestMl2SubnetsV2(test_plugin.TestSubnetsV2,
Ml2PluginV2TestCase):
def test_delete_subnet_race_with_dhcp_port_creation(self):
def test_l3_cleanup_on_net_delete(self):
l3plugin = manager.NeutronManager.get_service_plugins().get(
- service_constants.L3_ROUTER_NAT)
+ p_const.L3_ROUTER_NAT)
kwargs = {'arg_list': (external_net.EXTERNAL,),
external_net.EXTERNAL: True}
with self.network(**kwargs) as n:
ctx = context.get_admin_context()
plugin = manager.NeutronManager.get_plugin()
l3plugin = manager.NeutronManager.get_service_plugins().get(
- service_constants.L3_ROUTER_NAT)
+ p_const.L3_ROUTER_NAT)
with contextlib.nested(
self.port(),
mock.patch.object(l3plugin, 'disassociate_floatingips'),
def test_disassociate_floatingips_do_notify_returns_nothing(self):
ctx = context.get_admin_context()
l3plugin = manager.NeutronManager.get_service_plugins().get(
- service_constants.L3_ROUTER_NAT)
+ p_const.L3_ROUTER_NAT)
with self.port() as port:
port_id = port['port']['id']
def test_concurrent_csnat_port_delete(self):
plugin = manager.NeutronManager.get_service_plugins()[
- service_constants.L3_ROUTER_NAT]
+ p_const.L3_ROUTER_NAT]
r = plugin.create_router(
self.context,
{'router': {'name': 'router', 'admin_state_up': True}})
Code Review / openstack-build / neutron-build.git / commitdiff