Merge lp:~allenap/maas/narrow-port-range--bug-1352923 into lp:~maas-committers/maas/trunk

[Gavin Panella]

Bind the region's RPC endpoints to a narrow range of ports between 5250 and 5259 inclusive.

[Mike Pontillo (mpontillo)]

I had a quick chat with Gavin to answer some of my ignorant questions about this MP.

Q: Why do we need so many ports?!
A: Because each cluster is responsible for connecting to every regiond process. (each RPC endpoint is listed in JSON format if you hit <SERVER_URL>/MAAS/rpc/. (clusters periodically refresh this)

Q: Why can't the cluster just connect to any random regiond?
A: Because any arbitrary regiond needs to be able to use this control channel to talk to any arbitrary clusterd. Any individual regiond has no way to directly contact any arbitrary clusterd otherwise.

That being the case, I think this change looks good. I noted a minor issue with an 'except' clause below, and possibly a more robust test case.

review: Approve

[Gavin Panella (allenap)]

Thanks dudes!

I've also added a little bit of documentation.

=== modified file 'INSTALL.txt'
--- INSTALL.txt	2014-11-10 16:39:12 +0000
+++ INSTALL.txt	2015-05-22 16:26:39 +0000
@@ -323,5 +323,20 @@
   https://www.symantec.com/business/support/index?page=content&id=HOWTO6019
 
 
+Traffic between the region contoller and cluster controllers
+------------------------------------------------------------
+
+* Each cluster controller must be able to:
+
+  * Initiate TCP connections (for HTTP) to each region controller on
+    port 80 or port 5240, the choice of which depends on the setting of
+    ``MAAS_URL``.
+
+  * Initiate TCP connections (for RPC) to each region controller between
+    port 5250 and 5259 inclusive. This permits up to 10 ``maas-regiond``
+    processes on each region controller host. At present this is not
+    configurable.
+
+
 Once everything is set up and running, you are ready to :doc:`start
 enlisting nodes <nodes>`

=== modified file 'src/maasserver/rpc/regionservice.py'
--- src/maasserver/rpc/regionservice.py	2015-05-07 18:14:38 +0000
+++ src/maasserver/rpc/regionservice.py	2015-05-22 16:26:39 +0000
@@ -464,7 +464,10 @@
     This is a service - in the Twisted sense - that exposes the
     ``Region`` protocol on a port.
 
-    :ivar endpoints: The endpoints on which to listen.
+    :ivar endpoints: The endpoints on which to listen, as a list of lists.
+        Only one endpoint in each nested list will be bound (they will be
+        tried in order until the first success). In this way it is possible to
+        specify, say, a range of ports, but only bind one of them.
     :ivar ports: The opened :py:class:`IListeningPort`s.
     :ivar connections: Maps :class:`Region` connections to clusters.
     :ivar waiters: Maps cluster idents to callers waiting for a connection.
@@ -478,7 +481,10 @@
 
     def __init__(self):
         super(RegionService, self).__init__()
-        self.endpoints = [TCP4ServerEndpoint(reactor, 0)]
+        self.endpoints = [
+            [TCP4ServerEndpoint(reactor, port)
+             for port in xrange(5250, 5260)],
+        ]
         self.connections = defaultdict(set)
         self.waiters = defaultdict(set)
         self.factory = Factory.forProtocol(RegionServer)
@@ -529,15 +535,37 @@
             elif result.check(defer.CancelledError):
                 pass  # Ignore.
             else:
-                log.err(result)
+                log.err(result, "RegionServer endpoint failed to listen.")
+
+    @inlineCallbacks
+    def _bindFirst(self, endpoints, factory):
+        """Return the first endpoint to successfully listen.
+
+        :param endpoints: A sized iterable of `IStreamServerEndpoint`.
+        :param factory: A protocol factory.
+
+        :return: A `Deferred` yielding a :class:`twisted.internet.tcp.Port` or
+            the error encountered when trying to listen on the last of the
+            given endpoints.
+        """
+        assert len(endpoints) > 0, "No endpoint options specified."
+        last = len(endpoints) - 1
+        for index, endpoint in enumerate(endpoints):
+            try:
+                port = yield endpoint.listen(factory)
+            except:
+                if index == last:
+                    raise
+            else:
+                returnValue(port)
 
     @asynchronous
     def startService(self):
         """Start listening on an ephemeral port."""
         super(RegionService, self).startService()
         self.starting = defer.DeferredList(
-            (endpoint.listen(self.factory) for endpoint in self.endpoints),
-            consumeErrors=True)
+            (self._bindFirst(endpoint_options, self.factory)
+             for endpoint_options in self.endpoints))
 
         def log_failure(failure):
             if failure.check(defer.CancelledError):

=== modified file 'src/maasserver/rpc/testing/fixtures.py'
--- src/maasserver/rpc/testing/fixtures.py	2015-05-16 06:21:20 +0000
+++ src/maasserver/rpc/testing/fixtures.py	2015-05-22 16:26:39 +0000
@@ -272,7 +272,7 @@
         assert isinstance(self.rpc.endpoints, list)
         # Patch a fake UNIX endpoint in to the RPC service.
         endpoint = endpoints.UNIXServerEndpoint(reactor, self.sockfile)
-        self.monkey.add_patch(self.rpc, "endpoints", [endpoint])
+        self.monkey.add_patch(self.rpc, "endpoints", [[endpoint]])
 
         # The RPC service uses a defaultdict(set) to manage connections, but
         # let's check those assumptions.

=== modified file 'src/maasserver/rpc/tests/test_regionservice.py'
--- src/maasserver/rpc/tests/test_regionservice.py	2015-05-07 18:14:38 +0000
+++ src/maasserver/rpc/tests/test_regionservice.py	2015-05-22 16:26:39 +0000
@@ -83,6 +83,7 @@
     ANY,
     call,
     Mock,
+    sentinel,
 )
 import netaddr
 from provisioningserver.network import discover_networks
@@ -144,8 +145,10 @@
     HasLength,
     Is,
     IsInstance,
+    MatchesAll,
     MatchesListwise,
     MatchesStructure,
+    Not,
 )
 from twisted.application.service import Service
 from twisted.internet import (
@@ -160,6 +163,7 @@
     inlineCallbacks,
     succeed,
 )
+from twisted.internet.endpoints import TCP4ServerEndpoint
 from twisted.internet.error import ConnectionClosed
 from twisted.internet.interfaces import IStreamServerEndpoint
 from twisted.internet.protocol import Factory
@@ -167,6 +171,7 @@
 from twisted.protocols import amp
 from twisted.python import log
 from twisted.python.failure import Failure
+from twisted.python.reflect import fullyQualifiedName
 from zope.interface.verify import verifyObject
 
 
@@ -1223,7 +1228,8 @@
         self.assertThat(service.connections, IsInstance(defaultdict))
         self.assertThat(service.connections.default_factory, Is(set))
         self.assertThat(
-            service.endpoints, AllMatch(Provides(IStreamServerEndpoint)))
+            service.endpoints, AllMatch(
+                AllMatch(Provides(IStreamServerEndpoint))))
         self.assertThat(service.factory, IsInstance(Factory))
         self.assertThat(service.factory.protocol, Equals(RegionServer))
 
@@ -1274,8 +1280,8 @@
         service = RegionService()
 
         # Return an inert Deferred from the listen() call.
-        endpoints = self.patch(service, "endpoints", [Mock()])
-        endpoints[0].listen.return_value = Deferred()
+        endpoints = self.patch(service, "endpoints", [[Mock()]])
+        endpoints[0][0].listen.return_value = Deferred()
 
         service.startService()
         self.assertThat(service.starting, IsInstance(Deferred))
@@ -1296,28 +1302,112 @@
 
         # Ensure that endpoint.listen fails with a obvious error.
         exception = ValueError("This is not the messiah.")
-        endpoints = self.patch(service, "endpoints", [Mock()])
-        endpoints[0].listen.return_value = fail(exception)
-
-        err_calls = []
-        self.patch(log, "err", err_calls.append)
-
-        err_calls_expected = [
+        endpoints = self.patch(service, "endpoints", [[Mock()]])
+        endpoints[0][0].listen.return_value = fail(exception)
+
+        logged_failures = []
+        self.patch(log, "msg", (
+            lambda failure, **kw: logged_failures.append(failure)))
+
+        logged_failures_expected = [
             AfterPreprocessing(
                 (lambda failure: failure.value),
                 Is(exception)),
         ]
 
         yield service.startService()
-        self.assertThat(err_calls, MatchesListwise(err_calls_expected))
+        self.assertThat(
+            logged_failures, MatchesListwise(logged_failures_expected))
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test_start_up_binds_first_of_endpoint_options(self):
+        service = RegionService()
+
+        endpoint_1 = Mock()
+        endpoint_1.listen.return_value = succeed(sentinel.port1)
+        endpoint_2 = Mock()
+        endpoint_2.listen.return_value = succeed(sentinel.port2)
+        service.endpoints = [[endpoint_1, endpoint_2]]
+
+        yield service.startService()
+
+        self.assertThat(service.ports, Equals([sentinel.port1]))
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test_start_up_binds_first_of_real_endpoint_options(self):
+        service = RegionService()
+
+        # endpoint_1.listen(...) will bind to a random high-numbered port.
+        endpoint_1 = TCP4ServerEndpoint(reactor, 0)
+        # endpoint_2.listen(...), if attempted, will crash because only root
+        # (or a user with explicit capabilities) can do stuff like that. It's
+        # a reasonable assumption that the user running these tests is not
+        # root, but we'll check the port number later too to be sure.
+        endpoint_2 = TCP4ServerEndpoint(reactor, 1)
+
+        service.endpoints = [[endpoint_1, endpoint_2]]
+
+        yield service.startService()
+        self.addCleanup(wait_for_reactor(service.stopService))
+
+        # A single port has been bound.
+        self.assertThat(service.ports, MatchesAll(
+            HasLength(1), AllMatch(IsInstance(tcp.Port))))
+
+        # The port is not listening on port 1; i.e. a belt-n-braces check that
+        # endpoint_2 was not used.
+        [port] = service.ports
+        self.assertThat(port.getHost().port, Not(Equals(1)))
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test_start_up_binds_first_successful_of_endpoint_options(self):
+        service = RegionService()
+
+        endpoint_broken = Mock()
+        endpoint_broken.listen.return_value = fail(factory.make_exception())
+        endpoint_okay = Mock()
+        endpoint_okay.listen.return_value = succeed(sentinel.port)
+        service.endpoints = [[endpoint_broken, endpoint_okay]]
+
+        yield service.startService()
+
+        self.assertThat(service.ports, Equals([sentinel.port]))
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test_start_up_logs_failure_if_all_endpoint_options_fail(self):
+        service = RegionService()
+
+        error_1 = factory.make_exception_type()
+        error_2 = factory.make_exception_type()
+
+        endpoint_1 = Mock()
+        endpoint_1.listen.return_value = fail(error_1())
+        endpoint_2 = Mock()
+        endpoint_2.listen.return_value = fail(error_2())
+        service.endpoints = [[endpoint_1, endpoint_2]]
+
+        with TwistedLoggerFixture() as logger:
+            yield service.startService()
+
+        self.assertDocTestMatches(
+            """\
+            RegionServer endpoint failed to listen.
+            Traceback (most recent call last):
+            Failure: %s:
+            """ % fullyQualifiedName(error_2),
+            logger.output)
 
     @wait_for_reactor
     def test_stopping_cancels_startup(self):
         service = RegionService()
 
         # Return an inert Deferred from the listen() call.
-        endpoints = self.patch(service, "endpoints", [Mock()])
-        endpoints[0].listen.return_value = Deferred()
+        endpoints = self.patch(service, "endpoints", [[Mock()]])
+        endpoints[0][0].listen.return_value = Deferred()
 
         service.startService()
         service.stopService()
@@ -1391,8 +1481,8 @@
 
         # Ensure that endpoint.listen fails with a obvious error.
         exception = ValueError("This is a very naughty boy.")
-        endpoints = self.patch(service, "endpoints", [Mock()])
-        endpoints[0].listen.return_value = fail(exception)
+        endpoints = self.patch(service, "endpoints", [[Mock()]])
+        endpoints[0][0].listen.return_value = fail(exception)
         # Suppress logged messages.
         self.patch(log.theLogPublisher, "observers", [])