Merge pull request #46 from universalcore/feature/issue-46-sync-error…

…-handling

Ensure all errors handled when syncing apps

consular/cli.py

@@ -54,7 +54,6 @@ def main(scheme, host, port,
          sync_interval, purge, logfile, debug, timeout,
          fallback, fallback_timeout):  # pragma: no cover
     from consular.main import Consular
-    from twisted.internet.task import LoopingCall
     from twisted.internet import reactor
     from twisted.python import log
 
@@ -72,8 +71,7 @@ def main(scheme, host, port,
     consular.register_marathon_event_callback(events_url)
 
     if sync_interval > 0:
-        lc = LoopingCall(consular.sync_apps, purge)
-        lc.start(sync_interval, now=True)
+        consular.schedule_sync(sync_interval, purge)
 
     consular.run(host, port)
     reactor.run()

consular/main.py

@@ -5,6 +5,7 @@
 from twisted.internet import reactor
 from twisted.web import server
 from twisted.internet.defer import succeed, inlineCallbacks, returnValue
+from twisted.internet.task import LoopingCall
 from twisted.web.http import NOT_FOUND
 from twisted.python import log
 
@@ -108,6 +109,34 @@ def run(self, host, port):
         site.debug = self._debug
         self.clock.listenTCP(port, site, interface=host)
 
+    def schedule_sync(self, interval, purge=False):
+        """
+        Schedule a recurring sync of apps, starting after this method is
+        called.
+
+        :param float interval:
+            The number of seconds between syncs.
+        :param bool purge:
+            Whether to purge old apps after each sync.
+        :return:
+            A tuple of the LoopingCall object and the deferred created when it
+            was started.
+        """
+        lc = LoopingCall(self._try_sync_apps, purge)
+        lc.clock = self.clock
+        return (lc, lc.start(interval, now=True))
+
+    @inlineCallbacks
+    def _try_sync_apps(self, purge=False):
+        """
+        Sync the apps, catching and logging any exception that occurs.
+        """
+        try:
+            yield self.sync_apps(purge)
+        except Exception as e:
+            # TODO: More specialised exception handling.
+            log.msg('Error syncing apps: %s' % e)
+
     @inlineCallbacks
     def register_marathon_event_callback(self, events_url):
         """

consular/tests/test_main.py

@@ -6,6 +6,7 @@
 from twisted.internet import reactor
 from twisted.internet.defer import (
     inlineCallbacks, DeferredQueue, Deferred, succeed)
+from twisted.internet.task import Clock
 from twisted.web.client import HTTPConnectionPool
 from twisted.python import log
 
@@ -47,6 +48,7 @@ def setUp(self):
             'test'
         )
         self.consular.set_debug(True)
+        self.consular.clock = Clock()
 
         # spin up a site so we can test it, pretty sure Klein has better
         # ways of doing this but they're not documented anywhere.
@@ -267,6 +269,78 @@ def test_TASK_KILLED(self):
             'status': 'ok'
         })
 
+    @inlineCallbacks
+    def test_schedule_sync(self):
+        """
+        When Consular is set to schedule syncs, a sync should occur right away
+        and further syncs should occur after the correct delay.
+        """
+        lc, d = self.consular.schedule_sync(1)
+
+        self.assertTrue(lc.running)
+
+        # Consular should do the first sync right away
+        request = yield self.requests.get()
+        self.assertEqual(request['method'], 'GET')
+        self.assertEqual(
+            request['url'],
+            'http://localhost:8080/v2/apps')
+
+        # Return no apps... let's make this quick
+        request['deferred'].callback(
+            FakeResponse(200, [], json.dumps({'apps': []})))
+
+        # Advance the clock for the next sync
+        self.consular.clock.advance(1)
+
+        request = yield self.requests.get()
+        self.assertEqual(request['method'], 'GET')
+        self.assertEqual(
+            request['url'],
+            'http://localhost:8080/v2/apps')
+
+        request['deferred'].callback(
+            FakeResponse(200, [], json.dumps({'apps': []})))
+
+        lc.stop()
+        yield d
+
+    @inlineCallbacks
+    def test_schedule_sync_handles_server_errors(self):
+        """
+        When Consular is set to schedule syncs, syncs should not be interrupted
+        due to errors in previously scheduled syncs.
+        """
+        lc, d = self.consular.schedule_sync(1)
+
+        self.assertTrue(lc.running)
+
+        # Consular should do the first sync right away
+        request = yield self.requests.get()
+        self.assertEqual(request['method'], 'GET')
+        self.assertEqual(
+            request['url'],
+            'http://localhost:8080/v2/apps')
+
+        # Return a server error.
+        request['deferred'].callback(FakeResponse(500, [], 'Server error'))
+
+        # Advance the clock for the next sync
+        self.consular.clock.advance(1)
+
+        # The next sync should happen regardless of the previous server error
+        request = yield self.requests.get()
+        self.assertEqual(request['method'], 'GET')
+        self.assertEqual(
+            request['url'],
+            'http://localhost:8080/v2/apps')
+
+        request['deferred'].callback(
+            FakeResponse(200, [], json.dumps({'apps': []})))
+
+        lc.stop()
+        yield d
+
     @inlineCallbacks
     def test_register_with_marathon(self):
         d = self.consular.register_marathon_event_callback(