Merge ~blake-rouse/maas:fix-rack-rpc-conn into maas:master

[Blake Rouse]

Fixes LP: #1727073 - [2.3] rackd — 12% connected to region controllers.

Fix rackd RPC connections to only connect once and in parallel. Fix regiond RPC output to not include IP addresses that duplicate across regions. Fix precentage calculation for rackd connections.

All the code changes in the clusterservice are covered by the test_clusterservice. The try_connections needs covered so that was added to the tests.

I have tested this across a 2 region+rack controllers, with both having a configured virbr0 with IP 192.168.122.1.

[Andres Rodriguez (andreserl)]

codewise looks good, just have a couple of suggestions inline. Are those possible to do?

review: Needs Information

[Blake Rouse (blake-rouse)]

Okay its ready for another review. Logging has been updated.

[Andres Rodriguez (andreserl)]

review: Approve

[MAAS Lander (maas-lander)]

LANDING
-b fix-rack-rpc-conn lp:~blake-rouse/maas into -b master lp:~maas-committers/maas

STATUS: FAILED BUILD
LOG: http://maas-ci-jenkins.internal:8080/job/maas/job/branch-tester/496/consoleText

diff --git a/src/maasserver/models/node.py b/src/maasserver/models/node.py
index e6078b5..6699d81 100644
--- a/src/maasserver/models/node.py
+++ b/src/maasserver/models/node.py
@@ -5116,7 +5116,7 @@ class RackController(Controller):
                 Service.objects.update_service_for(
                     self, "rackd", SERVICE_STATUS.DEGRADED,
                     "{:.0%} connected to region controllers.".format(
-                        percentage))
+                        1.0 - percentage))
 
     def get_image_sync_status(self, boot_images=None):
         """Return the status of the boot image import process."""
diff --git a/src/maasserver/models/tests/test_node.py b/src/maasserver/models/tests/test_node.py
index c4bcb65..c3e5b09 100644
--- a/src/maasserver/models/tests/test_node.py
+++ b/src/maasserver/models/tests/test_node.py
@@ -9999,7 +9999,7 @@ class TestRackController(MAASTransactionServerTestCase):
             MatchesStructure.byEquality(
                 status=SERVICE_STATUS.DEGRADED, status_info=(
                     "{:.0%} connected to region controllers.".format(
-                        percentage))))
+                        1.0 - percentage))))
 
     fake_images = [
         {
diff --git a/src/maasserver/rpc/regionservice.py b/src/maasserver/rpc/regionservice.py
index 340a1be..5798823 100644
--- a/src/maasserver/rpc/regionservice.py
+++ b/src/maasserver/rpc/regionservice.py
@@ -1255,15 +1255,31 @@ class RegionAdvertising:
         Each tuple corresponds to somewhere an event-loop is listening
         within the whole region. The `name` is the event-loop name.
         """
+        # Each regiond might be running a local bridge that duplicates the
+        # same IP address across region controllers. Each region controller
+        # must output a set of unique of IP addresses, to prevent the rack
+        # controller from connecting to a different region controller then
+        # the rack controller was expecting to be connecting to.
+        def _unique_to_region(address, region, regions):
+            for region_obj in regions:
+                if region_obj != region:
+                    for process in region_obj.processes.all():
+                        for endpoint in process.endpoints.all():
+                            if endpoint.address == address:
+                                return False
+            return True
+
         regions = RegionController.objects.all()
         regions = regions.prefetch_related("processes", "processes__endpoints")
         all_endpoints = []
         for region_obj in regions:
             for process in region_obj.processes.all():
                 for endpoint in process.endpoints.all():
-                    all_endpoints.append((
-                        "%s:pid=%d" % (region_obj.hostname, process.pid),
-                        endpoint.address, endpoint.port))
+                    if _unique_to_region(
+                            endpoint.address, region_obj, regions):
+                        all_endpoints.append((
+                            "%s:pid=%d" % (region_obj.hostname, process.pid),
+                            endpoint.address, endpoint.port))
         return all_endpoints
 
     @classmethod
diff --git a/src/maasserver/rpc/tests/test_regionservice.py b/src/maasserver/rpc/tests/test_regionservice.py
index b3caf43..88c1a03 100644
--- a/src/maasserver/rpc/tests/test_regionservice.py
+++ b/src/maasserver/rpc/tests/test_regionservice.py
@@ -33,6 +33,7 @@ from maasserver.models import (
     RackController,
     RegionController,
     RegionControllerProcess,
+    RegionControllerProcessEndpoint,
     RegionRackRPCConnection,
     Service as ServiceModel,
     timestampedmodel,
@@ -1639,6 +1640,42 @@ class TestRegionAdvertising(MAASServerTestCase):
         ]
         self.assertItemsEqual(expected, advertising.dump())
 
+    def test_dump_doesnt_duplicate_ips_across_region_controllers(self):
+        duplicate_address = factory.make_ipv4_address()
+        addresses = {
+            (factory.make_ipv4_address(), factory.pick_port()),
+            (factory.make_ipv4_address(), factory.pick_port()),
+        }
+        advertising = RegionAdvertising.promote()
+        advertising.update(
+            addresses.union({(duplicate_address, factory.pick_port()), }))
+
+        other_addresses = {
+            (factory.make_ipv4_address(), factory.pick_port()),
+            (factory.make_ipv4_address(), factory.pick_port()),
+        }
+        other_region = factory.make_Node(node_type=NODE_TYPE.REGION_CONTROLLER)
+        other_process = RegionControllerProcess.objects.create(
+            region=other_region, pid=randint(1, 1000))
+        for address, port in other_addresses:
+            RegionControllerProcessEndpoint.objects.create(
+                process=other_process, address=address, port=port)
+        RegionControllerProcessEndpoint.objects.create(
+            process=other_process,
+            address=duplicate_address,
+            port=factory.pick_port())
+
+        expected = [
+            ("%s:pid=%d" % (gethostname(), os.getpid()), addr, port)
+            for (addr, port) in addresses
+        ] + [
+            ("%s:pid=%d" % (
+                other_region.hostname, other_process.pid),
+             addr, port)
+            for (addr, port) in other_addresses
+        ]
+        self.assertItemsEqual(expected, advertising.dump())
+
     def test__adds_connection_and_removes_connection(self):
         advertising = RegionAdvertising.promote()
         process = advertising.getRegionProcess()
diff --git a/src/provisioningserver/rpc/clusterservice.py b/src/provisioningserver/rpc/clusterservice.py
index d3e5413..946f516 100644
--- a/src/provisioningserver/rpc/clusterservice.py
+++ b/src/provisioningserver/rpc/clusterservice.py
@@ -116,6 +116,7 @@ from twisted import web
 from twisted.application.internet import TimerService
 from twisted.internet import reactor
 from twisted.internet.defer import (
+    DeferredList,
     inlineCallbacks,
     maybeDeferred,
     returnValue,
@@ -841,6 +842,8 @@ class ClusterClient(Cluster):
             log.msg("Event-loop '%s' authenticated." % self.ident)
             registered = yield self.registerRackWithRegion()
             if registered:
+                if self.eventloop in self.service.try_connections:
+                    del self.service.try_connections[self.eventloop]
                 self.service.connections[self.eventloop] = self
                 self.ready.set(self.eventloop)
             else:
@@ -992,6 +995,8 @@ class ClusterClientService(TimerService, object):
         super(ClusterClientService, self).__init__(
             self._calculate_interval(None, None), self._tryUpdate)
         self.connections = {}
+        self.try_connections = {}
+        self._previous_work = (None, None)
         self.clock = reactor
 
         # XXX jtv 2014-09-23, bug=1372767: Fix
@@ -1229,6 +1234,48 @@ class ClusterClientService(TimerService, object):
             ]
             for name, addresses in eventloops.items()
         }
+
+        drop, connect = self._calculate_work(eventloops)
+
+        # Log fully connected only once. If that state changes then log
+        # it again. This prevents flooding the log with the same message when
+        # the state of the connections has not changed.
+        prev_work, self._previous_work = self._previous_work, (drop, connect)
+        if len(drop) == 0 and len(connect) == 0:
+            if prev_work != (drop, connect) and len(eventloops) > 0:
+                controllers = {
+                    eventloop.split(':')[0]
+                    for eventloop, _ in eventloops.items()
+                }
+                log.msg(
+                    "Fully connected to all %d event-loops on all %d "
+                    "region controllers (%s)." % (
+                        len(eventloops), len(controllers),
+                        ', '.join(controllers)))
+
+        # Drop all connections at once, as the are no longer required.
+        if len(drop) > 0:
+            log.msg("Dropping connections to event-loops: %s" % (
+                ', '.join(drop.keys())))
+            yield DeferredList([
+                maybeDeferred(self._drop_connection, connection)
+                for eventloop, connections in drop.items()
+                for connection in connections
+            ], consumeErrors=True)
+
+        # Make all the new connections to each endpoint at the same time.
+        if len(connect) > 0:
+            log.msg("Making connections to event-loops: %s" % (
+                ', '.join(connect.keys())))
+            yield DeferredList([
+                self._make_connections(eventloop, addresses)
+                for eventloop, addresses in connect.items()
+            ], consumeErrors=True)
+
+    def _calculate_work(self, eventloops):
+        """Calculate the work that needs to be performed for reconnection."""
+        drop, connect = {}, {}
+
         # Drop connections to event-loops that no longer include one of
         # this cluster's established connections among its advertised
         # endpoints. This is most likely to have happened because of
@@ -1240,23 +1287,20 @@ class ClusterClientService(TimerService, object):
             if eventloop in self.connections:
                 connection = self.connections[eventloop]
                 if connection.address not in addresses:
-                    yield self._drop_connection(connection)
+                    drop[eventloop] = [connection]
+            if eventloop in self.try_connections:
+                connection = self.try_connections[eventloop]
+                if connection.address not in addresses:
+                    drop[eventloop] = [connection]
+
         # Create new connections to event-loops that the cluster does
-        # not yet have a connection to. Try each advertised endpoint
-        # (address) in turn until one of them bites.
+        # not yet have a connection to.
         for eventloop, addresses in eventloops.items():
-            if eventloop not in self.connections:
-                for address in addresses:
-                    try:
-                        yield self._make_connection(eventloop, address)
-                    except ConnectError as error:
-                        host, port = address
-                        log.msg("Event-loop %s (%s:%d): %s" % (
-                            eventloop, host, port, error))
-                    except:
-                        log.err(None, "Failure making new RPC connection.")
-                    else:
-                        break
+            if ((eventloop not in self.connections and
+                    eventloop not in self.try_connections) or
+                    eventloop in drop):
+                connect[eventloop] = addresses
+
         # Remove connections to event-loops that are no longer
         # advertised by the RPC info view. Most likely this means that
         # the process in which the event-loop is no longer running, but
@@ -1265,7 +1309,32 @@ class ClusterClientService(TimerService, object):
         for eventloop in self.connections:
             if eventloop not in eventloops:
                 connection = self.connections[eventloop]
-                yield self._drop_connection(connection)
+                drop[eventloop] = [connection]
+        for eventloop in self.try_connections:
+            if eventloop not in eventloops:
+                connection = self.try_connections[eventloop]
+                drop[eventloop] = [connection]
+
+        return drop, connect
+
+    @inlineCallbacks
+    def _make_connections(self, eventloop, addresses):
+        """Connect to `eventloop` using all `addresses`."""
+        for address in addresses:
+            try:
+                connection = yield self._make_connection(eventloop, address)
+            except ConnectError as error:
+                host, port = address
+                log.msg("Event-loop %s (%s:%d): %s" % (
+                    eventloop, host, port, error))
+            except:
+                host, port = address
+                log.err(None, (
+                    "Failure with event-loop %s (%s:%d)" % (
+                        eventloop, host, port)))
+            else:
+                self.try_connections[eventloop] = connection
+                break
 
     def _make_connection(self, eventloop, address):
         """Connect to `eventloop` at `address`."""
@@ -1284,6 +1353,9 @@ class ClusterClientService(TimerService, object):
         If this is the last connection that was keeping rackd connected to
         a regiond then dhcpd and dhcpd6 services will be turned off.
         """
+        if eventloop in self.try_connections:
+            if self.try_connections[eventloop] is connection:
+                del self.try_connections[eventloop]
         if eventloop in self.connections:
             if self.connections[eventloop] is connection:
                 del self.connections[eventloop]
@@ -1299,7 +1371,7 @@ class ClusterClientService(TimerService, object):
                 dhcp_v6.off()
                 stopping_services.append("dhcpd6")
             if len(stopping_services) > 0:
-                maaslog.error(
+                log.msg(
                     "Lost all connections to region controllers. "
                     "Stopping service(s) %s." % ",".join(stopping_services))
                 service_monitor.ensureServices()
diff --git a/src/provisioningserver/rpc/tests/test_clusterservice.py b/src/provisioningserver/rpc/tests/test_clusterservice.py
index 8553d02..9a7c85d 100644
--- a/src/provisioningserver/rpc/tests/test_clusterservice.py
+++ b/src/provisioningserver/rpc/tests/test_clusterservice.py
@@ -636,7 +636,9 @@ class TestClusterClientService(MAASTestCase):
     @inlineCallbacks
     def test__update_connections_initially(self):
         service = ClusterClientService(Clock())
+        mock_client = Mock()
         _make_connection = self.patch(service, "_make_connection")
+        _make_connection.side_effect = lambda *args: succeed(mock_client)
         _drop_connection = self.patch(service, "_drop_connection")
 
         info = json.loads(self.example_rpc_info_view_response.decode("ascii"))
@@ -650,10 +652,40 @@ class TestClusterClientService(MAASTestCase):
         self.assertItemsEqual(
             _make_connection_expected,
             _make_connection.call_args_list)
+        self.assertEquals({
+            "host1:pid=1001": mock_client,
+            "host1:pid=2002": mock_client,
+            "host2:pid=3003": mock_client,
+        }, service.try_connections)
 
         self.assertEqual([], _drop_connection.mock_calls)
 
     @inlineCallbacks
+    def test__update_connections_logs_fully_connected(self):
+        service = ClusterClientService(Clock())
+        eventloops = {
+            "region1:123": [("::ffff:127.0.0.1", 1234)],
+            "region1:124": [("::ffff:127.0.0.1", 1235)],
+            "region2:123": [("::ffff:127.0.0.2", 1234)],
+            "region2:124": [("::ffff:127.0.0.2", 1235)],
+        }
+        for eventloop, addresses in eventloops.items():
+            for address in addresses:
+                client = Mock()
+                client.address = address
+                service.connections[eventloop] = client
+
+        logger = self.useFixture(TwistedLoggerFixture())
+
+        yield service._update_connections(eventloops)
+        # Second call should not add it to the log.
+        yield service._update_connections(eventloops)
+
+        self.assertEqual(
+            "Fully connected to all 4 event-loops on all 2 region "
+            "controllers (region2, region1).", logger.dump())
+
+    @inlineCallbacks
     def test__update_connections_connect_error_is_logged_tersely(self):
         service = ClusterClientService(Clock())
         _make_connection = self.patch(service, "_make_connection")
@@ -669,6 +701,8 @@ class TestClusterClientService(MAASTestCase):
             MockCalledOnceWith("an-event-loop", ("::ffff:127.0.0.1", 1234)))
 
         self.assertEqual(
+            "Making connections to event-loops: an-event-loop\n"
+            "---\n"
             "Event-loop an-event-loop (::ffff:127.0.0.1:1234): Connection "
             "was refused by other side.", logger.dump())
 
@@ -689,7 +723,9 @@ class TestClusterClientService(MAASTestCase):
 
         self.assertDocTestMatches(
             """\
-            Failure making new RPC connection.
+            Making connections to event-loops: an-event-loop
+            ---
+            Failure with event-loop an-event-loop (::ffff:127.0.0.1:1234)
             Traceback (most recent call last):
             ...
             builtins.RuntimeError: Something went wrong.
@@ -797,6 +833,15 @@ class TestClusterClientService(MAASTestCase):
             connection.transport.loseConnection,
             MockCalledOnceWith())
 
+    def test__remove_connection_removes_from_try_connections(self):
+        service = make_inert_client_service()
+        service.startService()
+        endpoint = Mock()
+        connection = Mock()
+        service.try_connections[endpoint] = connection
+        service.remove_connection(endpoint, connection)
+        self.assertThat(service.try_connections, Equals({}))
+
     def test__remove_connection_removes_from_connections(self):
         service = make_inert_client_service()
         service.startService()
@@ -1063,6 +1108,7 @@ class TestClusterClient(MAASTestCase):
 
     def test_connecting(self):
         client = self.make_running_client()
+        client.service.try_connections[client.eventloop] = client
         self.patch_authenticate_for_success(client)
         self.patch_register_for_success(client)
         self.assertEqual(client.service.connections, {})
@@ -1076,6 +1122,7 @@ class TestClusterClient(MAASTestCase):
         self.assertTrue(extract_result(wait_for_authenticated))
         # ready has been set with the name of the event-loop.
         self.assertEqual(client.eventloop, extract_result(wait_for_ready))
+        self.assertEqual(client.service.try_connections, {})
         self.assertEqual(
             client.service.connections,
             {client.eventloop: client})