Merge ~mpontillo/maas:beaconing--use-hints-for-interface-update into maas:master

[Mike Pontillo]

Send topology hints to update_interfaces()

 * Add code to convert topology hints to a compact JSON format.
 * Start beaconing as soon as interfaces are known; don't
   wait for interface monitoring settings. This means
   beaconing will happen on all interfaces, all the time.
   (Without regard for network discovery settings.)
 * Fix bug that caused update_interfaces() to be called
   with create_fabrics=True when it should have been False.
 * Wait for beaconing to complete before updating interfaces.
 * Drive-by fix for incorrect line in .gitignore which
   erroneously covered django16_south_maas19.tar.gz.

[Blake Rouse (blake-rouse)]

Looks good. Just one question that I am not going to block you on. But if its the case that a region controller will never run beaconing? The question is why? Seems like it should.

review: Approve

[Mike Pontillo (mpontillo)]

Thanks for the review. The NetworksMonitoringService always runs beaconing, regardless of whether it's on the rack or the region. (The line you pointed out was in the tests; the tests for the region didn't need much change, since the region service simply calls update_interfaces() rather than going through the RPC layer.)

[MAAS Lander (maas-lander)]

LANDING
-b beaconing--use-hints-for-interface-update lp:~mpontillo/maas into -b master lp:~maas-committers/maas

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

[MAAS Lander (maas-lander)]

LANDING
-b beaconing--use-hints-for-interface-update lp:~mpontillo/maas into -b master lp:~maas-committers/maas

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

[Mike Pontillo (mpontillo)]

Fixed merge conflict with the .gitignore file, which slipped in when another branch landed.

[MAAS Lander (maas-lander)]

LANDING
-b beaconing--use-hints-for-interface-update lp:~mpontillo/maas into -b master lp:~maas-committers/maas

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

diff --git a/.gitignore b/.gitignore
index 20d7b98..714a000 100644
--- a/.gitignore
+++ b/.gitignore
@@ -19,6 +19,7 @@
 /.noseids
 /.run
 /.run-e2e
+/*.orig.tar.gz
 /bin
 /build
 /build_pkg
@@ -57,4 +58,3 @@ __pycache__
 *~
 \#*\#
 .#*
-*.tar.gz
diff --git a/src/maasserver/models/node.py b/src/maasserver/models/node.py
index 127669b..389f84c 100644
--- a/src/maasserver/models/node.py
+++ b/src/maasserver/models/node.py
@@ -4657,7 +4657,8 @@ class Controller(Node):
     @with_connection
     @synchronised(locks.rack_registration)
     @transactional
-    def update_interfaces(self, interfaces, create_fabrics=True):
+    def update_interfaces(
+            self, interfaces, topology_hints=None, create_fabrics=True):
         """Update the interfaces attached to the controller.
 
         :param interfaces: Interfaces dictionary that was parsed from
diff --git a/src/maasserver/regiondservices/networks_monitoring.py b/src/maasserver/regiondservices/networks_monitoring.py
index 9427bef..9cf313a 100644
--- a/src/maasserver/regiondservices/networks_monitoring.py
+++ b/src/maasserver/regiondservices/networks_monitoring.py
@@ -23,9 +23,10 @@ class RegionNetworksMonitoringService(NetworksMonitoringService):
         """Get interface monitoring state from the region."""
         return deferToDatabase(self.getInterfaceMonitoringStateFromDatabase)
 
-    def recordInterfaces(self, interfaces):
+    def recordInterfaces(self, interfaces, hints=None):
         """Record the interfaces information."""
-        return deferToDatabase(self.recordInterfacesIntoDatabase, interfaces)
+        return deferToDatabase(
+            self.recordInterfacesIntoDatabase, interfaces, hints)
 
     def reportNeighbours(self, neighbours):
         """Record the specified list of neighbours."""
@@ -42,10 +43,10 @@ class RegionNetworksMonitoringService(NetworksMonitoringService):
         return region_controller.get_discovery_state()
 
     @transactional
-    def recordInterfacesIntoDatabase(self, interfaces):
+    def recordInterfacesIntoDatabase(self, interfaces, hints):
         """Record the interfaces information."""
         region_controller = RegionController.objects.get_running_controller()
-        region_controller.update_interfaces(interfaces)
+        region_controller.update_interfaces(interfaces, hints)
 
     @transactional
     def recordNeighboursIntoDatabase(self, neighbours):
diff --git a/src/maasserver/regiondservices/tests/test_networks_monitoring.py b/src/maasserver/regiondservices/tests/test_networks_monitoring.py
index 62933ba..e49e34b 100644
--- a/src/maasserver/regiondservices/tests/test_networks_monitoring.py
+++ b/src/maasserver/regiondservices/tests/test_networks_monitoring.py
@@ -52,7 +52,8 @@ class TestRegionNetworksMonitoringService(MAASTransactionServerTestCase):
             }
         }
 
-        service = RegionNetworksMonitoringService(reactor)
+        service = RegionNetworksMonitoringService(
+            reactor, enable_beaconing=False)
         service.getInterfaces = lambda: succeed(interfaces)
 
         with FakeLogger("maas") as logger:
diff --git a/src/maasserver/rpc/rackcontrollers.py b/src/maasserver/rpc/rackcontrollers.py
index d7ed7e7..d80477c 100644
--- a/src/maasserver/rpc/rackcontrollers.py
+++ b/src/maasserver/rpc/rackcontrollers.py
@@ -222,10 +222,10 @@ def update_foreign_dhcp(system_id, interface_name, dhcp_ip=None):
 
 @synchronous
 @transactional
-def update_interfaces(system_id, interfaces):
+def update_interfaces(system_id, interfaces, topology_hints=None):
     """Update the interface definition on the rack controller."""
     rack_controller = RackController.objects.get(system_id=system_id)
-    rack_controller.update_interfaces(interfaces)
+    rack_controller.update_interfaces(interfaces, topology_hints)
 
 
 @synchronous
diff --git a/src/maasserver/rpc/regionservice.py b/src/maasserver/rpc/regionservice.py
index c6fc9d9..340a1be 100644
--- a/src/maasserver/rpc/regionservice.py
+++ b/src/maasserver/rpc/regionservice.py
@@ -404,14 +404,15 @@ class Region(RPCProtocol):
         return d
 
     @region.UpdateInterfaces.responder
-    def update_interfaces(self, system_id, interfaces):
+    def update_interfaces(self, system_id, interfaces, topology_hints=None):
         """update_interfaces()
 
         Implementation of
         :py:class:`~provisioningserver.rpc.region.UpdateInterfaces`.
         """
         d = deferToDatabase(
-            rackcontrollers.update_interfaces, system_id, interfaces)
+            rackcontrollers.update_interfaces, system_id, interfaces,
+            topology_hints=topology_hints)
         d.addCallback(lambda args: {})
         return d
 
@@ -616,8 +617,10 @@ class RegionServer(Region):
     def register(
             self, system_id, hostname, interfaces, url, nodegroup_uuid=None,
             beacon_support=False, version=None):
-        # If beacons is None, register in legacy mode.
-        create_fabrics = True if beacon_support else False
+        # Hold off on fabric creation if the remote controller
+        # supports beacons; it will happen later when UpdateInterfaces is
+        # called.
+        create_fabrics = False if beacon_support else True
         result = yield self._register(
             system_id, hostname, interfaces, url,
             nodegroup_uuid=nodegroup_uuid, create_fabrics=create_fabrics,
diff --git a/src/maasserver/rpc/tests/test_rackcontrollers.py b/src/maasserver/rpc/tests/test_rackcontrollers.py
index 1f40522..5392b32 100644
--- a/src/maasserver/rpc/tests/test_rackcontrollers.py
+++ b/src/maasserver/rpc/tests/test_rackcontrollers.py
@@ -479,7 +479,7 @@ class TestUpdateInterfaces(MAASServerTestCase):
         update_interfaces(rack_controller.system_id, sentinel.interfaces)
         self.assertThat(
             patched_update_interfaces,
-            MockCalledOnceWith(sentinel.interfaces))
+            MockCalledOnceWith(sentinel.interfaces, None))
 
 
 class TestReportNeighbours(MAASServerTestCase):
diff --git a/src/maasserver/rpc/tests/test_regionservice_calls.py b/src/maasserver/rpc/tests/test_regionservice_calls.py
index 1d1d497..209ee44 100644
--- a/src/maasserver/rpc/tests/test_regionservice_calls.py
+++ b/src/maasserver/rpc/tests/test_regionservice_calls.py
@@ -1182,7 +1182,8 @@ class TestRegionProtocol_UpdateInterfaces(MAASTransactionServerTestCase):
         self.assertThat(
             update_interfaces,
             MockCalledOnceWith(
-                params['system_id'], params['interfaces']))
+                params['system_id'], params['interfaces'],
+                topology_hints=None))
 
 
 class TestRegionProtocol_ReportNeighbours(MAASTestCase):
diff --git a/src/provisioningserver/rackdservices/networks_monitoring_service.py b/src/provisioningserver/rackdservices/networks_monitoring_service.py
index 1de76f2..d8f7e8c 100644
--- a/src/provisioningserver/rackdservices/networks_monitoring_service.py
+++ b/src/provisioningserver/rackdservices/networks_monitoring_service.py
@@ -8,6 +8,7 @@ __all__ = [
 ]
 
 from provisioningserver.logger import get_maas_logger
+from provisioningserver.rpc.exceptions import NoConnectionsAvailable
 from provisioningserver.rpc.region import (
     GetDiscoveryState,
     ReportMDNSEntries,
@@ -16,6 +17,8 @@ from provisioningserver.rpc.region import (
     UpdateInterfaces,
 )
 from provisioningserver.utils.services import NetworksMonitoringService
+from provisioningserver.utils.twisted import pause
+from twisted.internet.defer import inlineCallbacks
 
 
 maaslog = get_maas_logger("networks.monitor")
@@ -44,20 +47,21 @@ class RackNetworksMonitoringService(NetworksMonitoringService):
             d.addCallback(getState)
             return d
 
-    def recordInterfaces(self, interfaces):
+    @inlineCallbacks
+    def recordInterfaces(self, interfaces, hints=None):
         """Record the interfaces information to the region."""
-        def record(client):
-            # On first run perform a refresh
+        while self.running:
+            try:
+                client = yield self.clientService.getClientNow()
+            except NoConnectionsAvailable:
+                yield pause(1.0)
+                continue
             if self._recorded is None:
-                return client(RequestRackRefresh, system_id=client.localIdent)
-            else:
-                return client(
-                    UpdateInterfaces, system_id=client.localIdent,
-                    interfaces=interfaces)
-
-        d = self.clientService.getClientNow()
-        d.addCallback(record)
-        return d
+                yield client(RequestRackRefresh, system_id=client.localIdent)
+            yield client(
+                UpdateInterfaces, system_id=client.localIdent,
+                interfaces=interfaces, topology_hints=hints)
+            break
 
     def reportNeighbours(self, neighbours):
         """Report neighbour information to the region."""
diff --git a/src/provisioningserver/rackdservices/tests/test_networks_monitoring_service.py b/src/provisioningserver/rackdservices/tests/test_networks_monitoring_service.py
index d325125..2404cfa 100644
--- a/src/provisioningserver/rackdservices/tests/test_networks_monitoring_service.py
+++ b/src/provisioningserver/rackdservices/tests/test_networks_monitoring_service.py
@@ -5,8 +5,13 @@
 
 __all__ = []
 
+from unittest.mock import call
+
 from maastesting.factory import factory
-from maastesting.matchers import MockCalledOnceWith
+from maastesting.matchers import (
+    MockCalledOnceWith,
+    MockCallsMatch,
+)
 from maastesting.testcase import (
     MAASTestCase,
     MAASTwistedRunTest,
@@ -17,6 +22,7 @@ from provisioningserver.rackdservices.networks_monitoring_service import (
 )
 from provisioningserver.rpc import region
 from provisioningserver.rpc.testing import MockLiveClusterToRegionRPCFixture
+from provisioningserver.utils import services as services_module
 from twisted.internet.defer import (
     inlineCallbacks,
     maybeDeferred,
@@ -27,7 +33,7 @@ from twisted.internet.task import Clock
 
 class TestRackNetworksMonitoringService(MAASTestCase):
 
-    run_tests_with = MAASTwistedRunTest.make_factory(debug=False, timeout=5)
+    run_tests_with = MAASTwistedRunTest.make_factory(debug=True, timeout=5)
 
     @inlineCallbacks
     def test_runs_refresh_first_time(self):
@@ -37,10 +43,14 @@ class TestRackNetworksMonitoringService(MAASTestCase):
 
         rpc_service = services.getServiceNamed('rpc')
         service = RackNetworksMonitoringService(
-            rpc_service, Clock(), enable_monitoring=False)
+            rpc_service, Clock(), enable_monitoring=False,
+            enable_beaconing=False)
 
-        yield service.startService()
-        yield service.stopService()
+        yield maybeDeferred(service.startService)
+        # By stopping the interface_monitor first, we assure that the loop
+        # happens at least once before the service stops completely.
+        yield maybeDeferred(service.interface_monitor.stopService)
+        yield maybeDeferred(service.stopService)
 
         self.assertThat(
             protocol.RequestRackRefresh, MockCalledOnceWith(
@@ -64,7 +74,8 @@ class TestRackNetworksMonitoringService(MAASTestCase):
 
         rpc_service = services.getServiceNamed('rpc')
         service = RackNetworksMonitoringService(
-            rpc_service, Clock(), enable_monitoring=False)
+            rpc_service, Clock(), enable_monitoring=False,
+            enable_beaconing=False)
         service.getInterfaces = lambda: succeed(interfaces)
         # Put something in the cache. This tells recordInterfaces that refresh
         # has already run but the interfaces have changed thus they need to be
@@ -77,7 +88,57 @@ class TestRackNetworksMonitoringService(MAASTestCase):
         self.assertThat(
             protocol.UpdateInterfaces, MockCalledOnceWith(
                 protocol, system_id=rpc_service.getClient().localIdent,
-                interfaces=interfaces))
+                interfaces=interfaces, topology_hints=None))
+
+    @inlineCallbacks
+    def test_reports_interfaces_with_hints_if_beaconing_enabled(self):
+        fixture = self.useFixture(MockLiveClusterToRegionRPCFixture())
+        protocol, connecting = fixture.makeEventLoop(region.UpdateInterfaces)
+        # Don't actually wait for beaconing to complete.
+        pause_mock = self.patch(services_module, 'pause')
+        queue_mcast_mock = self.patch(
+            services_module.BeaconingSocketProtocol, 'queueMulticastBeaconing')
+        self.addCleanup((yield connecting))
+
+        interfaces = {
+            "eth0": {
+                "type": "physical",
+                "mac_address": factory.make_mac_address(),
+                "parents": [],
+                "links": [],
+                "enabled": True,
+            }
+        }
+
+        rpc_service = services.getServiceNamed('rpc')
+        service = RackNetworksMonitoringService(
+            rpc_service, Clock(), enable_monitoring=False,
+            enable_beaconing=True)
+        service.getInterfaces = lambda: succeed(interfaces)
+        # Put something in the cache. This tells recordInterfaces that refresh
+        # has already run but the interfaces have changed thus they need to be
+        # updated.
+        service._recorded = {}
+
+        service.startService()
+        yield service.stopService()
+
+        self.assertThat(
+            protocol.UpdateInterfaces, MockCalledOnceWith(
+                protocol, system_id=rpc_service.getClient().localIdent,
+                interfaces=interfaces, topology_hints=[]))
+        # The service should have sent out beacons, waited three seconds,
+        # solicited for more beacons, then waited another three seconds before
+        # deciding that beaconing is complete.
+        self.assertThat(pause_mock, MockCallsMatch(call(3.0), call(3.0)))
+        self.assertThat(
+            queue_mcast_mock, MockCallsMatch(
+                # Called when the service starts.
+                call(solicitation=True),
+                # Called three seconds later.
+                call(solicitation=True),
+                # Not called again when the service shuts down.
+            ))
 
     @inlineCallbacks
     def test_reports_neighbours_to_region(self):
@@ -87,7 +148,8 @@ class TestRackNetworksMonitoringService(MAASTestCase):
         self.addCleanup((yield connecting))
         rpc_service = services.getServiceNamed('rpc')
         service = RackNetworksMonitoringService(
-            rpc_service, Clock(), enable_monitoring=False)
+            rpc_service, Clock(), enable_monitoring=False,
+            enable_beaconing=False)
         neighbours = [{"ip": factory.make_ip_address()}]
         yield service.reportNeighbours(neighbours)
         self.assertThat(
@@ -103,7 +165,8 @@ class TestRackNetworksMonitoringService(MAASTestCase):
         self.addCleanup((yield connecting))
         rpc_service = services.getServiceNamed('rpc')
         service = RackNetworksMonitoringService(
-            rpc_service, Clock(), enable_monitoring=False)
+            rpc_service, Clock(), enable_monitoring=False,
+            enable_beaconing=False)
         mdns = [
             {
                 'interface': 'eth0',
@@ -126,7 +189,8 @@ class TestRackNetworksMonitoringService(MAASTestCase):
         rpc_service = services.getServiceNamed('rpc')
         reactor = Clock()
         service = RackNetworksMonitoringService(
-            rpc_service, reactor, enable_monitoring=False)
+            rpc_service, reactor, enable_monitoring=False,
+            enable_beaconing=False)
         protocol.GetDiscoveryState.return_value = {'interfaces': {}}
         # Put something in the cache. This tells recordInterfaces that refresh
         # has already run but the interfaces have changed thus they need to be
diff --git a/src/provisioningserver/rpc/region.py b/src/provisioningserver/rpc/region.py
index ab16e4b..58c43ab 100644
--- a/src/provisioningserver/rpc/region.py
+++ b/src/provisioningserver/rpc/region.py
@@ -460,6 +460,7 @@ class UpdateInterfaces(amp.Command):
     arguments = [
         (b'system_id', amp.Unicode()),
         (b'interfaces', StructureAsJSON()),
+        (b'topology_hints', StructureAsJSON(optional=True)),
     ]
     response = []
     errors = []
diff --git a/src/provisioningserver/utils/services.py b/src/provisioningserver/utils/services.py
index 5a43922..cb5c81b 100644
--- a/src/provisioningserver/utils/services.py
+++ b/src/provisioningserver/utils/services.py
@@ -28,7 +28,6 @@ from provisioningserver.logger import (
     get_maas_logger,
     LegacyLogger,
 )
-from provisioningserver.rpc.exceptions import NoConnectionsAvailable
 from provisioningserver.utils.beaconing import (
     age_out_uuid_queue,
     BEACON_IPV4_MULTICAST,
@@ -52,7 +51,7 @@ from provisioningserver.utils.shell import select_c_utf8_bytes_locale
 from provisioningserver.utils.twisted import (
     callOut,
     deferred,
-    suppress,
+    pause,
     terminateProcess,
 )
 from twisted.application.internet import TimerService
@@ -425,9 +424,15 @@ def join_ipv6_beacon_group(sock, ifindex):
         socket.inet_pton(socket.AF_INET6, BEACON_IPV6_MULTICAST) +
         struct.pack("I", ifindex)
     )
-    sock.setsockopt(
-        socket.IPPROTO_IPV6, socket.IPV6_JOIN_GROUP,
-        ipv6_join_sockopt_args)
+    try:
+        sock.setsockopt(
+            socket.IPPROTO_IPV6, socket.IPV6_JOIN_GROUP,
+            ipv6_join_sockopt_args)
+    except OSError:
+        # Do this on a best-effort basis. We might get an "Address already in
+        # use" error if the group is already joined, or (for whatever reason)
+        # it is not possible to join a multicast group using this interface.
+        pass
 
 
 def set_ipv6_multicast_loopback(sock, loopback):
@@ -472,6 +477,7 @@ class BeaconingSocketProtocol(DatagramProtocol):
         self.topology_hints = OrderedDict()
         self.listen_port = None
         self.mcast_requested = False
+        self.mcast_solicitation = False
         self.last_solicited_mcast = 0
         self._join_multicast_groups()
 
@@ -482,8 +488,9 @@ class BeaconingSocketProtocol(DatagramProtocol):
             verifyObject(IReactorMulticast, self.reactor)
         except DoesNotImplement:
             return
-        self.listen_port = self.reactor.listenMulticast(
-            self.port, self, interface=self.interface, listenMultiple=True)
+        if self.listen_port is None:
+            self.listen_port = self.reactor.listenMulticast(
+                self.port, self, interface=self.interface, listenMultiple=True)
         sock = self.transport.getHandle()
         if self.loopback is True:
             # This is only necessary for testing.
@@ -512,6 +519,34 @@ class BeaconingSocketProtocol(DatagramProtocol):
         self.interfaces = interfaces
         self._join_multicast_groups()
 
+    def getAllTopologyHints(self):
+        """Returns the set of unique topology hints."""
+        # When beaconing runs, hints attached to individual packets might
+        # come to the same conclusion about the implied fabric connectivity.
+        # Use a set to prevent the region from processing duplicate hints.
+        all_hints = set()
+        for hints in self.topology_hints.values():
+            all_hints |= hints
+        return all_hints
+
+    def getJSONTopologyHints(self):
+        """Returns all topology hints as a list of dictionaries.
+
+        This method is used for sending data via the RPC layer, so be cautious
+        when modifying. In addition, keys with no value are filtered out
+        of the resulting dictionary, so that the hints are smaller on the wire.
+        """
+        all_hints = self.getAllTopologyHints()
+        json_hints = [
+            {
+                key: value
+                for key, value in hint._asdict().items()
+                if value is not None
+            }
+            for hint in all_hints
+        ]
+        return json_hints
+
     def stopProtocol(self):
         super().stopProtocol()
         if self.listen_port is not None:
@@ -629,27 +664,36 @@ class BeaconingSocketProtocol(DatagramProtocol):
         reply = create_beacon_payload("advertisement", payload)
         self.send_beacon(reply, beacon.reply_address)
         if len(self.interfaces) > 0:
-            self.queueMulticastAdvertisement()
+            self.queueMulticastBeaconing()
 
-    def dequeueMulticastAdvertisement(self):
+    def dequeueMulticastBeaconing(self):
         """
         Callback to send multicast beacon advertisements.
 
         See `queueMulticastAdvertisement`, which schedules this method to run.
         """
         mtime = time.monotonic()
+        beacon_type = (
+            'solicitation' if self.mcast_solicitation else 'advertisement')
+        log.msg("Sending multicast beacon %ss." % beacon_type)
+        self.send_multicast_beacons(self.interfaces, beacon_type)
         self.last_solicited_mcast = mtime
         self.mcast_requested = False
-        log.msg("Sending multicast beacon advertisements.")
-        self.send_multicast_beacons(self.interfaces, 'advertisement')
+        self.mcast_solicitation = False
 
-    def queueMulticastAdvertisement(self):
+    def queueMulticastBeaconing(self, solicitation=False):
         """
         Requests that multicast advertisements be sent out on every interface.
 
         Ensures that advertisements will not be sent more than once every
         five seconds.
+
+        :param solicitation: If true, sends solicitations rather than
+            advertisements. Solicitations are used to initiate "full beaconing"
+            with peers; advertisements do not generate beacon replies.
         """
+        if solicitation is True:
+            self.mcast_solicitation = True
         if self.mcast_requested:
             # A multicast advertisement has been requested already.
             return
@@ -660,7 +704,7 @@ class BeaconingSocketProtocol(DatagramProtocol):
         else:
             timeout = max(mtime - self.last_solicited_mcast, 5)
         self.mcast_requested = True
-        self.reactor.callLater(timeout, self.dequeueMulticastAdvertisement)
+        self.reactor.callLater(timeout, self.dequeueMulticastBeaconing)
 
     def processTopologyHints(self, rx: ReceivedBeacon):
         """
@@ -692,9 +736,7 @@ class BeaconingSocketProtocol(DatagramProtocol):
             self.topology_hints[rx.uuid] = hints
             # XXX mpontillo 2017-08-07: temporary logging
             log.msg("New topology hints [%s]:\n%s" % (rx.uuid, pformat(hints)))
-            all_hints = set()
-            for hints in self.topology_hints.values():
-                all_hints |= hints
+            all_hints = self.getAllTopologyHints()
             log.msg("Topology hint summary:\n%s" % pformat(all_hints))
 
     def _add_remote_fabric_hints(self, hints, remote_ifinfo, rx):
@@ -905,7 +947,8 @@ class NetworksMonitoringService(MultiService, metaclass=ABCMeta):
 
     interval = timedelta(seconds=30).total_seconds()
 
-    def __init__(self, clock=None, enable_monitoring=True):
+    def __init__(
+            self, clock=None, enable_monitoring=True, enable_beaconing=True):
         # Order is very important here. First we set the clock to the passed-in
         # reactor, so that unit tests can fake out the clock if necessary.
         # Then we call super(). The superclass will set up the structures
@@ -916,9 +959,11 @@ class NetworksMonitoringService(MultiService, metaclass=ABCMeta):
         self.clock = clock
         super().__init__()
         self.enable_monitoring = enable_monitoring
+        self.enable_beaconing = enable_beaconing
         # The last successfully recorded interfaces.
         self._recorded = None
         self._monitored = frozenset()
+        self._beaconing = frozenset()
         self._monitoring_state = {}
         self._monitoring_mdns = False
         self._locked = False
@@ -935,33 +980,25 @@ class NetworksMonitoringService(MultiService, metaclass=ABCMeta):
         self.interface_monitor.setServiceParent(self)
         self.beaconing_protocol = None
 
+    @inlineCallbacks
     def updateInterfaces(self):
         """Update interfaces, catching and logging errors.
 
         This can be overridden by subclasses to conditionally update based on
         some external configuration.
         """
-        d = maybeDeferred(self._assumeSoleResponsibility)
-
-        def update(responsible):
-            if responsible:
-                d = maybeDeferred(self.getInterfaces)
-                d.addCallback(self._updateInterfaces)
-                return d
-
-        def failed(failure):
-            log.err(
-                failure,
-                "Failed to update and/or record network interface "
-                "configuration: %s" % failure.getErrorMessage())
-
-        d = d.addCallback(update)
-        # During the update, we might fail to get the interface monitoring
-        # state from the region. We can safely ignore this, as it will be
-        # retried shortly.
-        d.addErrback(suppress, NoConnectionsAvailable)
-        d.addErrback(failed)
-        return d
+        responsible = self._assumeSoleResponsibility()
+        if responsible:
+            interfaces = None
+            try:
+                interfaces = yield maybeDeferred(self.getInterfaces)
+                yield self._updateInterfaces(interfaces)
+            except BaseException as e:
+                msg = (
+                    "Failed to update and/or record network interface "
+                    "configuration: %s; interfaces: %r" % (e, interfaces)
+                )
+                log.err(None, msg)
 
     def getInterfaces(self):
         """Get the current network interfaces configuration.
@@ -978,7 +1015,7 @@ class NetworksMonitoringService(MultiService, metaclass=ABCMeta):
         """
 
     @abstractmethod
-    def recordInterfaces(self, interfaces):
+    def recordInterfaces(self, interfaces, hints=None):
         """Record the interfaces information.
 
         This MUST be overridden in subclasses.
@@ -1050,21 +1087,48 @@ class NetworksMonitoringService(MultiService, metaclass=ABCMeta):
             # If we were monitoring neighbours on any interfaces, we need to
             # stop the monitoring services.
             self._configureNetworkDiscovery({})
+            if self.beaconing_protocol is not None:
+                self._configureBeaconing({})
 
+    @inlineCallbacks
     def _updateInterfaces(self, interfaces):
         """Record `interfaces` if they've changed."""
         if interfaces != self._recorded:
-            d = maybeDeferred(self.recordInterfaces, interfaces)
+            hints = None
+            if self.enable_beaconing:
+                self._configureBeaconing(interfaces)
+                # Wait for beaconing to do its thing.
+                yield pause(3.0)
+                # Retry beacon soliciations, in case any packet loss occurred
+                # the first time.
+                self.beaconing_protocol.queueMulticastBeaconing(
+                    solicitation=True)
+                yield pause(3.0)
+                hints = self.beaconing_protocol.getJSONTopologyHints()
+            yield maybeDeferred(self.recordInterfaces, interfaces, hints)
             # Note: _interfacesRecorded() will reconfigure discovery after
             # recording the interfaces, so there is no need to call
             # _configureNetworkDiscovery() here.
-            d.addCallback(callOut, self._interfacesRecorded, interfaces)
-            return d
+            self._interfacesRecorded(interfaces)
         else:
             # If the interfaces didn't change, we still need to poll for
             # monitoring state changes.
-            d = maybeDeferred(self._configureNetworkDiscovery, interfaces)
-            return d
+            yield maybeDeferred(self._configureNetworkDiscovery, interfaces)
+
+    def _getInterfacesForBeaconing(self, interfaces: dict):
+        """Return the interfaces which will be used for beaconing.
+
+        :return: The set of interface names to run beaconing on.
+        """
+        # Don't beacon when running the test suite/dev env.
+        # In addition, if we don't own the lock, we should not be beaconing.
+        if is_dev_environment() or not self._locked or interfaces is None:
+            return set()
+        monitored_interfaces = {
+            ifname for ifname, ifdata in interfaces.items()
+            if ifdata['monitored'] is True
+        }
+        return monitored_interfaces
 
     def _getInterfacesForNeighbourDiscovery(
             self, interfaces: dict, monitoring_state: dict):
@@ -1227,6 +1291,32 @@ class NetworksMonitoringService(MultiService, metaclass=ABCMeta):
             # doesn't matter for mDNS discovery purposes.)
             pass
 
+    def _configureBeaconing(self, interfaces):
+        beaconing_interfaces = self._getInterfacesForBeaconing(interfaces)
+        # Calculate the difference between the sets. We need to know which
+        # interfaces were added and deleted (with respect to the interfaces we
+        # were already beaconing on).
+        new_interfaces = beaconing_interfaces.difference(self._beaconing)
+        deleted_interfaces = self._beaconing.difference(beaconing_interfaces)
+        if len(new_interfaces) > 0:
+            log.msg("Starting beaconing for interfaces: %r" % (new_interfaces))
+            self._startBeaconingServices(new_interfaces)
+        if len(deleted_interfaces) > 0:
+            log.msg(
+                "Stopping beaconing for interfaces: %r" % (deleted_interfaces))
+            self._stopBeaconingServices(deleted_interfaces)
+        self._beaconing = beaconing_interfaces
+        if self.beaconing_protocol is None:
+            self.beaconing_protocol = BeaconingSocketProtocol(
+                self.clock, interfaces=interfaces)
+        else:
+            self.beaconing_protocol.updateInterfaces(interfaces)
+        # If the interfaces have changed, perform beaconing again.
+        # An empty dictionary will be passed in when the service stops, so
+        # don't bother sending out beacons we won't reply to.
+        if len(interfaces) > 0:
+            self.beaconing_protocol.queueMulticastBeaconing(solicitation=True)
+
     def _configureNeighbourDiscovery(self, interfaces, monitoring_state):
         monitored_interfaces = self._getInterfacesForNeighbourDiscovery(
             interfaces, monitoring_state)
@@ -1239,26 +1329,15 @@ class NetworksMonitoringService(MultiService, metaclass=ABCMeta):
             log.msg("Starting neighbour discovery for interfaces: %r" % (
                 new_interfaces))
             self._startNeighbourDiscoveryServices(new_interfaces)
-            # XXX mpontillo 2017-07-12: for testing, just start beaconing
-            # services on all the interfaces enabled for active discovery.
-            self._startBeaconingServices(new_interfaces)
         if len(deleted_interfaces) > 0:
             log.msg(
                 "Stopping neighbour discovery for interfaces: %r" % (
                     deleted_interfaces))
             self._stopNeighbourDiscoveryServices(deleted_interfaces)
-            # XXX mpontillo 2017-07-12: this should be separately configured.
-            # (see similar comment in the 'start' path above.)
-            self._stopBeaconingServices(deleted_interfaces)
         self._monitored = monitored_interfaces
 
     def _interfacesRecorded(self, interfaces):
         """The given `interfaces` were recorded successfully."""
         self._recorded = interfaces
-        if self.beaconing_protocol is None:
-            self.beaconing_protocol = BeaconingSocketProtocol(
-                self.clock, interfaces=interfaces)
-        else:
-            self.beaconing_protocol.updateInterfaces(interfaces)
         if self.enable_monitoring is True:
             self._configureNetworkDiscovery(interfaces)
diff --git a/src/provisioningserver/utils/tests/test_services.py b/src/provisioningserver/utils/tests/test_services.py
index acfc60e..36090ea 100644
--- a/src/provisioningserver/utils/tests/test_services.py
+++ b/src/provisioningserver/utils/tests/test_services.py
@@ -76,8 +76,12 @@ from twisted.python.failure import Failure
 class StubNetworksMonitoringService(NetworksMonitoringService):
     """Concrete subclass for testing."""
 
-    def __init__(self, enable_monitoring=False, *args, **kwargs):
-        super().__init__(enable_monitoring=enable_monitoring, *args, **kwargs)
+    def __init__(
+            self, enable_monitoring=False, enable_beaconing=False,
+            *args, **kwargs):
+        super().__init__(
+            *args, enable_monitoring=enable_monitoring,
+            enable_beaconing=enable_beaconing, **kwargs)
         self.iterations = DeferredQueue()
         self.interfaces = []
         self.update_interface__calls = 0
@@ -91,7 +95,7 @@ class StubNetworksMonitoringService(NetworksMonitoringService):
         d.addBoth(self.iterations.put)
         return d
 
-    def recordInterfaces(self, interfaces):
+    def recordInterfaces(self, interfaces, hints=None):
         self.interfaces.append(interfaces)
 
     def reportNeighbours(self, neighbours):
@@ -221,13 +225,13 @@ class TestNetworksMonitoringService(MAASTestCase):
             # recordInterfaces is called the first time, as expected.
             recordInterfaces.reset_mock()
             yield service.updateInterfaces()
-            self.assertThat(recordInterfaces, MockCalledOnceWith({}))
+            self.assertThat(recordInterfaces, MockCalledOnceWith({}, None))
 
             # recordInterfaces is called the second time too; the service noted
             # that it crashed last time and knew to run it again.
             recordInterfaces.reset_mock()
             yield service.updateInterfaces()
-            self.assertThat(recordInterfaces, MockCalledOnceWith({}))
+            self.assertThat(recordInterfaces, MockCalledOnceWith({}, None))
 
             # recordInterfaces is NOT called the third time; the service noted
             # that the configuration had not changed.
@@ -1004,3 +1008,128 @@ class TestBeaconingSocketProtocol(SharedSecretTestCase):
         }
         self.assertThat(hints, Equals(expected_hints))
         yield protocol.stopProtocol()
+
+    @inlineCallbacks
+    def test__getJSONTopologyHints_converts_hints_to_dictionary(self):
+        # Note: Always use a random port for testing. (port=0)
+        protocol = BeaconingSocketProtocol(
+            reactor, port=0, process_incoming=False, loopback=True,
+            interface="::", debug=True)
+        # Don't try to send out any replies.
+        self.patch(services, 'create_beacon_payload')
+        self.patch(protocol, 'send_beacon')
+        # Need to generate a real UUID with the current time, so it doesn't
+        # get aged out.
+        uuid = str(uuid1())
+        # Make the protocol think we sent a beacon with this UUID already.
+        tx_mac = factory.make_mac_address()
+        fake_tx_beacon = FakeBeaconPayload(
+            uuid, ifname='eth1', mac=tx_mac, vid=100)
+        fake_rx_beacon = {
+            "source_ip": "127.0.0.1",
+            "source_port": 5240,
+            "destination_ip": "224.0.0.118",
+            "interface": "eth0",
+            "type": "solicitation",
+            "payload": fake_tx_beacon.payload
+        }
+        protocol.beaconReceived(fake_rx_beacon)
+        all_hints = protocol.getJSONTopologyHints()
+        expected_hints = [
+            # Note: since vid=None on the received beacon, we expect that
+            # the hint won't have a 'vid' field.
+            dict(
+                ifname='eth0', hint="on_remote_network",
+                related_ifname='eth1', related_vid=100,
+                related_mac=tx_mac),
+        ]
+        self.assertThat(all_hints, Equals(expected_hints))
+        yield protocol.stopProtocol()
+
+    @inlineCallbacks
+    def test__queues_multicast_beacon_soliciations_upon_request(self):
+        # Note: Always use a random port for testing. (port=0)
+        clock = Clock()
+        protocol = BeaconingSocketProtocol(
+            clock, port=0, process_incoming=False, loopback=True,
+            interface="::", debug=True)
+        # Don't try to send out any replies.
+        self.patch(services, 'create_beacon_payload')
+        send_mcast_mock = self.patch(protocol, 'send_multicast_beacons')
+        self.patch(protocol, 'send_beacon')
+        yield protocol.queueMulticastBeaconing(solicitation=True)
+        clock.advance(0)
+        self.assertThat(
+            send_mcast_mock, MockCalledOnceWith({}, 'solicitation'))
+
+    @inlineCallbacks
+    def test__multicasts_at_most_once_per_five_seconds(self):
+        # Note: Always use a random port for testing. (port=0)
+        clock = Clock()
+        protocol = BeaconingSocketProtocol(
+            clock, port=0, process_incoming=False, loopback=True,
+            interface="::", debug=True)
+        # Don't try to send out any replies.
+        self.patch(services, 'create_beacon_payload')
+        monotonic_mock = self.patch(services.time, 'monotonic')
+        send_mcast_mock = self.patch(protocol, 'send_multicast_beacons')
+        self.patch(protocol, 'send_beacon')
+        monotonic_mock.side_effect = [
+            # Initial queue
+            6,
+            # Initial dequeue
+            6,
+            # Second queue (hasn't yet been 5 seconds)
+            10,
+            # Third queue
+            11,
+            # Second dequeue
+            11,
+        ]
+        yield protocol.queueMulticastBeaconing()
+        clock.advance(0)
+        self.assertThat(
+            send_mcast_mock, MockCalledOnceWith({}, 'advertisement'))
+        send_mcast_mock.reset_mock()
+        yield protocol.queueMulticastBeaconing()
+        yield protocol.queueMulticastBeaconing(solicitation=True)
+        clock.advance(4.9)
+        self.assertThat(
+            send_mcast_mock, MockNotCalled())
+        clock.advance(0.1)
+        self.assertThat(
+            send_mcast_mock, MockCalledOnceWith({}, 'solicitation'))
+
+    @inlineCallbacks
+    def test__multiple_beacon_requests_coalesced(self):
+        # Note: Always use a random port for testing. (port=0)
+        clock = Clock()
+        protocol = BeaconingSocketProtocol(
+            clock, port=0, process_incoming=False, loopback=True,
+            interface="::", debug=True)
+        # Don't try to send out any replies.
+        self.patch(services, 'create_beacon_payload')
+        send_mcast_mock = self.patch(protocol, 'send_multicast_beacons')
+        self.patch(protocol, 'send_beacon')
+        yield protocol.queueMulticastBeaconing()
+        yield protocol.queueMulticastBeaconing()
+        clock.advance(5)
+        self.assertThat(
+            send_mcast_mock, MockCalledOnceWith({}, 'advertisement'))
+
+    @inlineCallbacks
+    def test__solicitation_wins_when_multiple_requests_queued(self):
+        # Note: Always use a random port for testing. (port=0)
+        clock = Clock()
+        protocol = BeaconingSocketProtocol(
+            clock, port=0, process_incoming=False, loopback=True,
+            interface="::", debug=True)
+        # Don't try to send out any replies.
+        self.patch(services, 'create_beacon_payload')
+        send_mcast_mock = self.patch(protocol, 'send_multicast_beacons')
+        self.patch(protocol, 'send_beacon')
+        yield protocol.queueMulticastBeaconing()
+        yield protocol.queueMulticastBeaconing(solicitation=True)
+        clock.advance(5)
+        self.assertThat(
+            send_mcast_mock, MockCalledOnceWith({}, 'solicitation'))