Merge lp:~allenap/maas/there-can-be-only-one--bug-1462320 into lp:~maas-committers/maas/trunk

[Gavin Panella]

Remove overlapping event-loop records where it's obvious they're outdated.

In addition, don't allow RegionAdvertisingService to crash when there's an error, log instead.

[Raphaël Badin (rvb)]

Looks good. Tested in my lab and it works as expected.

review: Approve

=== modified file 'src/maasserver/rpc/regionservice.py'
--- src/maasserver/rpc/regionservice.py	2015-06-02 15:50:27 +0000
+++ src/maasserver/rpc/regionservice.py	2015-06-05 13:23:18 +0000
@@ -698,7 +698,12 @@
 
     def __init__(self, interval=60):
         super(RegionAdvertisingService, self).__init__(
-            interval, deferToThread, self.update)
+            interval, self.try_update)
+
+    def try_update(self):
+        return deferToThread(self.update).addErrback(
+            log.err, "Failed to update this event-loop's advertisement; "
+            "%s's record may be out of date" % (eventloop.loop.name,))
 
     @asynchronous
     def startService(self):
@@ -784,9 +789,11 @@
         event-loop running in the same process, and deletes - garbage
         collects - old records related to any event-loop.
         """
+        addresses = frozenset(self._get_addresses())
         with closing(connection.cursor()) as cursor:
             self._do_delete(cursor)
-            self._do_insert(cursor)
+            self._do_delete_overlap(cursor, addresses)
+            self._do_insert(cursor, addresses)
             self._do_collect(cursor)
 
     @synchronous
@@ -881,13 +888,25 @@
     def _do_delete(self, cursor):
         cursor.execute(self._delete_statement, [eventloop.loop.name])
 
+    _delete_overlap_statement = "DELETE FROM eventloops WHERE "
+    _delete_overlap_values_statement = "(address = %s AND port = %s)"
+
+    def _do_delete_overlap(self, cursor, addresses):
+        statement, values = [], []
+        for addr, port in addresses:
+            statement.append(self._delete_overlap_values_statement)
+            values.extend([addr, port])
+        if len(statement) != 0:
+            statement = self._delete_overlap_statement + " OR ".join(statement)
+            cursor.execute(statement, values)
+
     _insert_statement = "INSERT INTO eventloops (name, address, port) VALUES "
     _insert_values_statement = "(%s, %s, %s)"
 
-    def _do_insert(self, cursor):
+    def _do_insert(self, cursor, addresses):
         name = eventloop.loop.name
         statement, values = [], []
-        for addr, port in self._get_addresses():
+        for addr, port in addresses:
             statement.append(self._insert_values_statement)
             values.extend([name, addr, port])
         if len(statement) != 0:

=== modified file 'src/maasserver/rpc/tests/test_regionservice.py'
--- src/maasserver/rpc/tests/test_regionservice.py	2015-06-02 15:50:27 +0000
+++ src/maasserver/rpc/tests/test_regionservice.py	2015-06-05 13:23:18 +0000
@@ -1718,7 +1718,25 @@
     def test_init(self):
         ras = RegionAdvertisingService()
         self.assertEqual(60, ras.step)
-        self.assertEqual((deferToThread, (ras.update,), {}), ras.call)
+        self.assertEqual((ras.try_update, (), {}), ras.call)
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test_try_update_logs_all_errors(self):
+        ras = RegionAdvertisingService()
+        error = factory.make_exception_type()
+        self.patch(ras, "update").side_effect = error
+        with TwistedLoggerFixture() as logger:
+            yield ras.try_update()
+        self.assertDocTestMatches(
+            """
+            Failed to update this event-loop's advertisement;
+              %s's record may be out of date
+            Traceback (most recent call last):
+            ...
+            maastesting.factory.TestException#...
+            """ % eventloop.loop.name,
+            logger.output)
 
     @wait_for_reactor
     def test_starting_and_stopping_the_service(self):
@@ -1926,6 +1944,70 @@
             [("vic", "192.168.1.1", 1111)],
             service.dump())
 
+    def patch_port(self, port):
+        getServiceNamed = self.patch(eventloop.services, "getServiceNamed")
+        getPort = getServiceNamed.return_value.getPort
+        getPort.side_effect = asynchronous(lambda: port)
+
+    def patch_addresses(self, addresses):
+        get_all_interface_addresses = self.patch(
+            regionservice, "get_all_interface_addresses")
+        get_all_interface_addresses.return_value = addresses
+
+    def test_update_deletes_overlapping_records(self):
+        service = RegionAdvertisingService()
+        service.prepare()
+
+        # The name of a fictional old event-loop.
+        example_name = factory.make_name("loop")
+        # Port and addresses on which the old and current loops are listening.
+        example_port = factory.pick_port()
+        example_addrs = {
+            factory.make_ipv4_address(),
+            factory.make_ipv4_address(),
+        }
+        self.patch_port(example_port)
+        self.patch_addresses(example_addrs)
+
+        # A non-overlapping host and port for the old event-loop.
+        for example_addr in iter(factory.make_ipv4_address, None):
+            if example_addr not in example_addrs:
+                example_non_overlapping_record = (
+                    example_name, example_addr, example_port)
+                break
+
+        # Populate the eventloops table with records for the old event-loop.
+        with closing(connection):
+            with closing(connection.cursor()) as cursor:
+                # Create the non-overlapping record.
+                cursor.execute("""\
+                  INSERT INTO eventloops (name, address, port, updated)
+                  VALUES (%s, %s, %s, DEFAULT)
+                """, example_non_overlapping_record)
+                # Create the overlapping records.
+                for example_addr in example_addrs:
+                    cursor.execute("""\
+                      INSERT INTO eventloops (name, address, port, updated)
+                      VALUES (%s, %s, %s, DEFAULT)
+                    """, [example_name, example_addr, example_port])
+
+        # All records for the old event-loop are present.
+        self.assertItemsEqual(
+            [example_non_overlapping_record] + [
+                (example_name, example_addr, example_port)
+                for example_addr in example_addrs
+            ],
+            service.dump())
+        # Updating replaces overlapping records with records for the current
+        # event-loop. The non-overlapping record remains.
+        service.update()
+        self.assertItemsEqual(
+            [example_non_overlapping_record] + [
+                (eventloop.loop.name, example_addr, example_port)
+                for example_addr in example_addrs
+            ],
+            service.dump())
+
     def test_dump(self):
         example_addresses = [
             (factory.make_ipv4_address(), factory.pick_port()),
@@ -1975,9 +2057,7 @@
         service = RegionAdvertisingService()
 
         example_port = factory.pick_port()
-        getServiceNamed = self.patch(eventloop.services, "getServiceNamed")
-        getPort = getServiceNamed.return_value.getPort
-        getPort.side_effect = asynchronous(lambda: example_port)
+        self.patch_port(example_port)
 
         example_ipv4_addrs = {
             factory.make_ipv4_address(),
@@ -1991,9 +2071,7 @@
             factory.pick_ip_in_network(netaddr.ip.IPV4_LINK_LOCAL),
             factory.pick_ip_in_network(netaddr.ip.IPV6_LINK_LOCAL),
         }
-        get_all_interface_addresses = self.patch(
-            regionservice, "get_all_interface_addresses")
-        get_all_interface_addresses.return_value = (
+        self.patch_addresses(
             example_ipv4_addrs | example_ipv6_addrs |
             example_link_local_addrs)
 
@@ -2003,8 +2081,12 @@
             [(addr, example_port) for addr in example_ipv4_addrs],
             service._get_addresses())
 
-        getServiceNamed.assert_called_once_with("rpc")
-        get_all_interface_addresses.assert_called_once_with()
+        self.assertThat(
+            eventloop.services.getServiceNamed,
+            MockCalledOnceWith("rpc"))
+        self.assertThat(
+            regionservice.get_all_interface_addresses,
+            MockCalledOnceWith())
 
     def test__get_addresses_when_rpc_down(self):
         service = RegionAdvertisingService()