Add fullstack cross-process port/ip address fixtures

We've had a series of bugs with resources that need
to be unique on the system across test runner
processes. Ports are used by neutron-server and the
OVS agent when run in native openflow mode. The function
that generates ports looks up random unused ports and
starts the service. However, it is raceful: By the time the
port is found to be unused and the service is started,
another test runner can pick the same random port.
With close to 65536 ports to choose from, the chance
for collision is low, but given enough test runs, it's
happened a non-trivial amount of times, and given that
a voting job needs a very low false-negative rate, we
need a more robust solution. The same applies to IP
addresses that are used by the OVS agent in tunneling
mode, and for the LB agent in all modes. With IP addresses,
we don't check if the IP address is used, we simply
pick a random address from a large pool, and again
we've seen a non-trivial amount of test failures.

The bugs referenced below had simple, short term solutions
applied but the bugs remain remain. This patch is a correct,
long term solution that doesn't rely on chance.

This patch adds a resource allocator that uses the disk
to persist allocations. Access to the disk is guarded
via a file lock. IP address, networks and ports fixtures
use an allocator internally.

Closes-Bug: #1551288
Closes-Bug: #1561248
Closes-Bug: #1560277
Change-Id: I46c0ca138b806759128462f8d44a5fab96a106d3

neutron/tests/common/exclusive_resources/ip_address.py

@@ -0,0 +1,41 @@
+# Copyright 2016 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import functools
+import random
+
+import netaddr
+
+from neutron.tests.common.exclusive_resources import resource_allocator
+
+
+def get_random_ip(low, high):
+    parent_range = netaddr.IPRange(low, high)
+    return str(random.choice(parent_range))
+
+
+class ExclusiveIPAddress(resource_allocator.ExclusiveResource):
+    """Allocate a unique ip address.
+
+    :ivar address: allocated ip address
+    :type address: netaddr.IPAddress
+    """
+
+    def __init__(self, low, high):
+        super(ExclusiveIPAddress, self).__init__(
+            'ip_addresses', functools.partial(get_random_ip, low, high))
+
+    def _setUp(self):
+        super(ExclusiveIPAddress, self)._setUp()
+        self.address = netaddr.IPAddress(self.resource)

neutron/tests/common/exclusive_resources/ip_network.py

@@ -0,0 +1,48 @@
+# Copyright 2016 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import functools
+
+import netaddr
+
+from neutron.tests.common.exclusive_resources import ip_address
+from neutron.tests.common.exclusive_resources import resource_allocator
+
+
+def _get_random_network(low, high, netmask):
+    ip = ip_address.get_random_ip(low, high)
+    return str(netaddr.IPNetwork("%s/%s" % (ip, netmask)).cidr)
+
+
+class ExclusiveIPNetwork(resource_allocator.ExclusiveResource):
+    """Allocate a non-overlapping ip network.
+
+    :ivar network: allocated ip network
+    :type network: netaddr.IPNetwork
+    """
+
+    def __init__(self, low, high, netmask):
+        super(ExclusiveIPNetwork, self).__init__(
+            'ip_networks',
+            functools.partial(_get_random_network, low, high, netmask),
+            self.is_valid)
+
+    def _setUp(self):
+        super(ExclusiveIPNetwork, self)._setUp()
+        self.network = netaddr.IPNetwork(self.resource)
+
+    def is_valid(self, new_resource, allocated_resources):
+        new_ipset = netaddr.IPSet([new_resource])
+        allocated_ipset = netaddr.IPSet(allocated_resources)
+        return new_ipset.isdisjoint(allocated_ipset)

neutron/tests/common/exclusive_resources/port.py

@@ -0,0 +1,35 @@
+# Copyright 2016 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import functools
+
+from neutron.tests.common.exclusive_resources import resource_allocator
+from neutron.tests.common import net_helpers
+
+
+class ExclusivePort(resource_allocator.ExclusiveResource):
+    """Allocate a unique port for a specific protocol.
+
+    :ivar port: allocated port
+    :type port: int
+    """
+
+    def __init__(self, protocol):
+        super(ExclusivePort, self).__init__(
+            'ports',
+            functools.partial(net_helpers.get_free_namespace_port, protocol))
+
+    def _setUp(self):
+        super(ExclusivePort, self)._setUp()
+        self.port = self.resource

neutron/tests/common/exclusive_resources/resource_allocator.py

@@ -0,0 +1,105 @@
+# Copyright 2016 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import os
+
+import fixtures
+
+from neutron.common import utils
+
+
+MAX_ATTEMPTS = 100
+TMP_DIR = '/tmp/neutron_exclusive_resources/'
+
+
+class ExclusiveResource(fixtures.Fixture):
+    def __init__(self, resource_name, allocator_function, validator=None):
+        self.ra = ResourceAllocator(
+            resource_name, allocator_function, validator)
+
+    def _setUp(self):
+        self.resource = self.ra.allocate()
+        self.addCleanup(self.ra.release, self.resource)
+
+
+class ResourceAllocator(object):
+    """ResourceAllocator persists cross-process allocations of a resource.
+
+    Allocations are persisted to a file determined by the 'resource_name',
+    and are allocated via an allocator_function. The public interface
+    (allocate and release) are guarded by a file lock. The intention
+    is to allow atomic, cross-process allocation of shared resources
+    such as ports and IP addresses. For usages of this class, please see
+    ExclusiveIPAddress and its functional tests.
+
+    Note that this class doesn't maintain in-memory state, and multiple
+    instances of it may be initialized and used. A pool of resources
+    is identified solely by the 'resource_name' argument.
+    """
+    def __init__(self, resource_name, allocator_function, validator=None):
+        """Initialize a resource allocator.
+
+        :param resource_name: A unique identifier for a pool of resources.
+        :param allocator_function: A function with no parameters that generates
+                                   a resource.
+        :param validator: An optional function that accepts a resource and an
+                          existing pool and returns if the generated resource
+                          is valid.
+        """
+        def is_valid(new_resource, allocated_resources):
+            return new_resource not in allocated_resources
+
+        self._allocator_function = allocator_function
+        self._state_file_path = os.path.join(TMP_DIR, resource_name)
+        self._validator = validator if validator else is_valid
+
+    @utils.synchronized('resource_allocator', external=True, lock_path='/tmp')
+    def allocate(self):
+        allocations = self._get_allocations()
+
+        for i in range(MAX_ATTEMPTS):
+            resource = str(self._allocator_function())
+            if self._validator(resource, allocations):
+                allocations.add(resource)
+                self._write_allocations(allocations)
+                return resource
+
+        raise ValueError(
+            'Could not allocate a new resource in %s from pool %s' %
+            (self.__class__.__name__, allocations))
+
+    @utils.synchronized('resource_allocator', external=True, lock_path='/tmp')
+    def release(self, resource):
+        allocations = self._get_allocations()
+        allocations.discard(resource)
+        if allocations:
+            self._write_allocations(allocations)
+        else:  # Clean up the file if we're releasing the last allocation
+            os.remove(self._state_file_path)
+
+    def _get_allocations(self):
+        utils.ensure_dir(TMP_DIR)
+
+        try:
+            with open(self._state_file_path, 'r') as allocations_file:
+                contents = allocations_file.read()
+        except IOError:
+            contents = None
+
+        # If the file was empty, we want to return an empty set, not {''}
+        return set(contents.split(',')) if contents else set()
+
+    def _write_allocations(self, allocations):
+        with open(self._state_file_path, 'w') as allocations_file:
+            allocations_file.write(','.join(allocations))

neutron/tests/fullstack/resources/config.py

@@ -20,18 +20,8 @@ from neutron.common import constants
 from neutron.plugins.ml2.extensions import qos as qos_ext
 from neutron.tests import base
 from neutron.tests.common import config_fixtures
+from neutron.tests.common.exclusive_resources import port
 from neutron.tests.common import helpers as c_helpers
-from neutron.tests.common import net_helpers
-
-
-def _generate_port():
-    """Get a free TCP port from the Operating System and return it.
-
-    This might fail if some other process occupies this port after this
-    function finished but before the neutron-server process started.
-    """
-    return str(net_helpers.get_free_namespace_port(
-        constants.PROTO_NAME_TCP))
 
 
 class ConfigFixture(fixtures.Fixture):
@@ -73,7 +63,6 @@ class NeutronConfigFixture(ConfigFixture):
                 'host': self._generate_host(),
                 'state_path': self._generate_state_path(self.temp_dir),
                 'lock_path': '$state_path/lock',
-                'bind_port': _generate_port(),
                 'api_paste_config': self._generate_api_paste(),
                 'policy_file': self._generate_policy_json(),
                 'core_plugin': 'neutron.plugins.ml2.plugin.Ml2Plugin',
@@ -93,6 +82,13 @@ class NeutronConfigFixture(ConfigFixture):
             }
         })
 
+    def _setUp(self):
+        self.config['DEFAULT'].update({
+            'bind_port': self.useFixture(
+                port.ExclusivePort(constants.PROTO_NAME_TCP)).port
+        })
+        super(NeutronConfigFixture, self)._setUp()
+
     def _generate_host(self):
         return base.get_rand_name(prefix='host-')
 
@@ -163,10 +159,6 @@ class OVSConfigFixture(ConfigFixture):
             }
         })
 
-        if self.config['ovs']['of_interface'] == 'native':
-            self.config['ovs'].update({
-                'of_listen_port': _generate_port()})
-
         if self.tunneling_enabled:
             self.config['agent'].update({
                 'tunnel_types': self.env_desc.network_type})
@@ -181,6 +173,14 @@ class OVSConfigFixture(ConfigFixture):
         if env_desc.qos:
             self.config['agent']['extensions'] = 'qos'
 
+    def _setUp(self):
+        if self.config['ovs']['of_interface'] == 'native':
+            self.config['ovs'].update({
+                'of_listen_port': self.useFixture(
+                    port.ExclusivePort(constants.PROTO_NAME_TCP)).port
+            })
+        super(OVSConfigFixture, self)._setUp()
+
     def _generate_bridge_mappings(self):
         return 'physnet1:%s' % base.get_rand_device_name(prefix='br-eth')
 

neutron/tests/fullstack/resources/environment.py

@@ -12,10 +12,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-import random
-
 import fixtures
-import netaddr
 from neutronclient.common import exceptions as nc_exc
 from oslo_config import cfg
 
@@ -25,6 +22,8 @@ from neutron.common import constants
 from neutron.common import utils as common_utils
 from neutron.plugins.ml2.drivers.linuxbridge.agent import \
     linuxbridge_neutron_agent as lb_agent
+from neutron.tests.common.exclusive_resources import ip_address
+from neutron.tests.common.exclusive_resources import ip_network
 from neutron.tests.common import net_helpers
 from neutron.tests.fullstack.resources import config
 from neutron.tests.fullstack.resources import process
@@ -85,8 +84,6 @@ class Host(fixtures.Fixture):
         self.host_desc = host_desc
         self.test_name = test_name
         self.neutron_config = neutron_config
-        # Use reserved class E addresses
-        self.local_ip = self.allocate_local_ip()
         self.central_data_bridge = central_data_bridge
         self.central_external_bridge = central_external_bridge
         self.host_namespace = None
@@ -96,6 +93,8 @@ class Host(fixtures.Fixture):
         self.network_bridges = {}
 
     def _setUp(self):
+        self.local_ip = self.allocate_local_ip()
+
         if self.host_desc.l2_agent_type == constants.AGENT_TYPE_OVS:
             self.setup_host_with_ovs_agent()
         elif self.host_desc.l2_agent_type == constants.AGENT_TYPE_LINUXBRIDGE:
@@ -211,11 +210,13 @@ class Host(fixtures.Fixture):
 
     def allocate_local_ip(self):
         if not self.env_desc.network_range:
-            return self.get_random_ip('240.0.0.1', '240.255.255.254')
-        return self.get_random_ip(
-            str(self.env_desc.network_range[2]),
-            str(self.env_desc.network_range[-2])
-        )
+            return str(self.useFixture(
+                ip_address.ExclusiveIPAddress(
+                    '240.0.0.1', '240.255.255.254')).address)
+        return str(self.useFixture(
+            ip_address.ExclusiveIPAddress(
+                str(self.env_desc.network_range[2]),
+                str(self.env_desc.network_range[-2]))).address)
 
     def get_bridge(self, network_id):
         if "ovs" in self.agents.keys():
@@ -233,11 +234,6 @@ class Host(fixtures.Fixture):
                 self.network_bridges[network_id] = bridge
         return bridge
 
-    @staticmethod
-    def get_random_ip(low, high):
-        parent_range = netaddr.IPRange(low, high)
-        return str(random.choice(parent_range))
-
     @property
     def hostname(self):
         return self.neutron_config.config.DEFAULT.host
@@ -365,10 +361,6 @@ class Environment(fixtures.Fixture):
         # address is fine for them
         for desc in self.hosts_desc:
             if desc.l2_agent_type == constants.AGENT_TYPE_LINUXBRIDGE:
-                return self.get_random_network(
-                    "240.0.0.0", "240.255.255.255", "24")
-
-    @staticmethod
-    def get_random_network(low, high, netmask):
-        ip = Host.get_random_ip(low, high)
-        return netaddr.IPNetwork("%s/%s" % (ip, netmask))
+                return self.useFixture(
+                    ip_network.ExclusiveIPNetwork(
+                        "240.0.0.0", "240.255.255.255", "24")).network

neutron/tests/functional/tests/common/exclusive_resources/test_ip_address.py

@@ -0,0 +1,29 @@
+# Copyright 2016 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import netaddr
+
+from neutron.tests import base
+from neutron.tests.common.exclusive_resources import ip_address
+
+
+class TestExclusiveIPAddress(base.DietTestCase):
+    def test_ip_address(self):
+        address_1 = self.useFixture(
+            ip_address.ExclusiveIPAddress('10.0.0.1', '10.0.0.2')).address
+        address_2 = self.useFixture(
+            ip_address.ExclusiveIPAddress('10.0.0.1', '10.0.0.2')).address
+
+        self.assertIsInstance(address_1, netaddr.IPAddress)
+        self.assertNotEqual(address_1, address_2)

neutron/tests/functional/tests/common/exclusive_resources/test_ip_network.py

@@ -0,0 +1,32 @@
+# Copyright 2016 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import netaddr
+
+from neutron.tests import base
+from neutron.tests.common.exclusive_resources import ip_network
+
+
+class TestExclusiveIPNetwork(base.DietTestCase):
+    def test_ip_network(self):
+        network_1 = self.useFixture(
+            ip_network.ExclusiveIPNetwork(
+                '240.0.0.1', '240.255.255.254', '24')).network
+        network_2 = self.useFixture(
+            ip_network.ExclusiveIPNetwork(
+                '240.0.0.1', '240.255.255.254', '24')).network
+
+        self.assertIsInstance(network_1, netaddr.IPNetwork)
+        self.assertEqual(network_1.cidr, network_1)
+        self.assertNotEqual(network_1, network_2)

neutron/tests/functional/tests/common/exclusive_resources/test_port.py

@@ -0,0 +1,28 @@
+# Copyright 2016 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from neutron.common import constants
+from neutron.tests import base
+from neutron.tests.common.exclusive_resources import port
+
+
+class TestExclusivePort(base.DietTestCase):
+    def test_port(self):
+        port_1 = self.useFixture(port.ExclusivePort(
+            constants.PROTO_NAME_TCP)).port
+        port_2 = self.useFixture(port.ExclusivePort(
+            constants.PROTO_NAME_TCP)).port
+
+        self.assertIsInstance(port_1, str)
+        self.assertNotEqual(port_1, port_2)

neutron/tests/functional/tests/common/exclusive_resources/test_resource_allocator.py

@@ -0,0 +1,61 @@
+# Copyright 2016 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import os
+
+import testtools
+
+from neutron.common import utils
+from neutron.tests import base
+from neutron.tests.common.exclusive_resources import resource_allocator
+
+
+def safe_remove_file(file_path):
+    try:
+        os.remove(file_path)
+    except OSError:
+        pass
+
+
+class TestResourceAllocator(base.DietTestCase):
+    def setUp(self):
+        super(TestResourceAllocator, self).setUp()
+        self.ra = resource_allocator.ResourceAllocator(
+            utils.get_random_string(6), lambda: 42)
+        self.addCleanup(safe_remove_file, self.ra._state_file_path)
+
+    def test_allocate_and_release(self):
+        # Assert that we can allocate a resource
+        resource = self.ra.allocate()
+        self.assertEqual('42', resource)
+
+        # Assert that we cannot allocate any more resources, since we're
+        # using an allocator that always returns the same value
+        with testtools.ExpectedException(ValueError):
+            self.ra.allocate()
+
+        # Assert that releasing the resource and allocating again works
+        self.ra.release(resource)
+        resource = self.ra.allocate()
+        self.assertEqual('42', resource)
+
+    def test_file_manipulation(self):
+        # The file should not be created until the first allocation
+        self.assertFalse(os.path.exists(self.ra._state_file_path))
+        resource = self.ra.allocate()
+        self.assertTrue(os.path.exists(self.ra._state_file_path))
+
+        # Releasing the last resource should delete the file
+        self.ra.release(resource)
+        self.assertFalse(os.path.exists(self.ra._state_file_path))