Allow specifying multiple Marathon addresses and failover to each sequentially

README.md

@@ -23,8 +23,8 @@ The ACME provider that most people are likely to use is [Let's Encrypt](https://
 
 ```
 > $ docker run --rm praekeltfoundation/marathon-acme:develop marathon-acme --help
-usage: marathon-acme [-h] [-a ACME] [-m MARATHON] [-l LB [LB ...]] [-g GROUP]
-                     [--listen LISTEN]
+usage: marathon-acme [-h] [-a ACME] [-m MARATHON [MARATHON ...]]
+                     [-l LB [LB ...]] [-g GROUP] [--listen LISTEN]
                      [--log-level {debug,info,warn,error,critical}]
                      storage-dir
 
@@ -37,7 +37,7 @@ optional arguments:
   -h, --help            show this help message and exit
   -a ACME, --acme ACME  The address for the ACME Directory Resource (default:
                         https://acme-v01.api.letsencrypt.org/directory)
-  -m MARATHON, --marathon MARATHON
+  -m MARATHON [MARATHON ...], --marathon MARATHON [MARATHON ...]
                         The address for the Marathon HTTP API (default:
                         http://marathon.mesos:8080)
   -l LB [LB ...], --lb LB [LB ...]
@@ -119,15 +119,15 @@ The `marathon-acme` Docker container can be configured either using command-line
 
 The environment variables available correspond to the CLI options as follows:
 
-| Environment variable      | CLI option    |
-|---------------------------|---------------|
-| `MARATHON_ACME_ACME`      | `--acme`      |
-| `MARATHON_ACME_MARATHON`  | `--marathon`  |
-| `MARATHON_ACME_LBS`*      | `--lb`        |
-| `MARATHON_ACME_GROUP`     | `--group`     |
-| `MARATHON_ACME_LOG_LEVEL` | `--log-level` |
+| Environment variable       | CLI option    |
+|----------------------------|---------------|
+| `MARATHON_ACME_ACME`       | `--acme`      |
+| `MARATHON_ACME_MARATHONS`* | `--marathon`  |
+| `MARATHON_ACME_LBS`*       | `--lb`        |
+| `MARATHON_ACME_GROUP`      | `--group`     |
+| `MARATHON_ACME_LOG_LEVEL`  | `--log-level` |
 
-\*Multiple load balancers can be set using multiple spaces-separated addresses.
+\*Multiple addresses should be space-separated.
 
 #### Volumes and ports
 The `marathon-acme` container defaults to the `/var/lib/marathon-acme` directory to store certificates and the ACME client private key. This is the path inside the container that should be mounted as a shared volume.

docker-entrypoint.sh

@@ -2,7 +2,7 @@
 
 # Options can be set using environment variables:
 # MARATHON_ACME_ACME:      --acme
-# MARATHON_ACME_MARATHON:  --marathon
+# MARATHON_ACME_MARATHONS:  --marathon
 # MARATHON_ACME_LBS:       --lb
 # MARATHON_ACME_GROUP:     --group
 # MARATHON_ACME_LOG_LEVEL: --log-level
@@ -13,7 +13,7 @@
 
 exec marathon-acme \
   ${MARATHON_ACME_ACME:+--acme "$MARATHON_ACME_ACME"} \
-  ${MARATHON_ACME_MARATHON:+--marathon "$MARATHON_ACME_MARATHON"} \
+  ${MARATHON_ACME_MARATHONS:+--marathon $MARATHON_ACME_MARATHONS} \
   ${MARATHON_ACME_LBS:+--lb $MARATHON_ACME_LBS} \
   ${MARATHON_ACME_GROUP:+--group "$MARATHON_ACME_GROUP"} \
   ${MARATHON_ACME_LOG_LEVEL:+--log-level "$MARATHON_ACME_LOG_LEVEL"} \

marathon_acme/cli.py

@@ -22,7 +22,7 @@
                          '(default: %(default)s)',
                     default=(
                         'https://acme-v01.api.letsencrypt.org/directory'))
-parser.add_argument('-m', '--marathon',
+parser.add_argument('-m', '--marathon', nargs='+',
                     help='The address for the Marathon HTTP API (default: '
                          '%(default)s)',
                     default='http://marathon.mesos:8080')
@@ -69,7 +69,7 @@ def main(reactor, raw_args=sys.argv[1:]):
 
 
 def create_marathon_acme(storage_dir, acme_directory,
-                         marathon_addr, mlb_addrs, group,
+                         marathon_addrs, mlb_addrs, group,
                          reactor):
     """
     Create a marathon-acme instance.
@@ -93,7 +93,7 @@ def create_marathon_acme(storage_dir, acme_directory,
     key = maybe_key(storage_path)
 
     return MarathonAcme(
-        MarathonClient(marathon_addr, reactor=reactor),
+        MarathonClient(marathon_addrs, reactor=reactor),
         group,
         DirectoryStore(certs_path),
         MarathonLbClient(mlb_addrs, reactor=reactor),

marathon_acme/clients.py

@@ -259,6 +259,44 @@ def sse_content(response, handler):
 
 class MarathonClient(JsonClient):
 
+    def __init__(self, endpoints, *args, **kwargs):
+        """
+        :param endpoints:
+            A priority-ordered list of Marathon endpoints. Each endpoint will
+            be tried one-by-one until the request succeeds or all endpoints
+            fail.
+        """
+        super(MarathonClient, self).__init__(*args, **kwargs)
+        self.endpoints = endpoints
+
+    def request(self, *args, **kwargs):
+        d = self._request(None, list(self.endpoints), *args, **kwargs)
+        d.addErrback(self._log_all_endpoints_failed)
+        return d
+
+    def _request(self, failure, endpoints, *args, **kwargs):
+        """
+        Recursively make requests to each endpoint in ``endpoints``.
+        """
+        # We've run out of endpoints, fail
+        if not endpoints:
+            return failure
+
+        endpoint = endpoints.pop(0)
+        d = super(MarathonClient, self).request(*args, url=endpoint, **kwargs)
+
+        # If something goes wrong, call ourselves again with the remaining
+        # endpoints
+        d.addErrback(self._request, endpoints, *args, **kwargs)
+        return d
+
+    def _log_all_endpoints_failed(self, failure):
+        # Just log an error so it is clear what has happened and return the
+        # final failure. Individual failures should have been logged via
+        # _log_request_error().
+        self.log.error('Failed to make a request to all Marathon endpoints')
+        return failure
+
     def get_json_field(self, field, **kwargs):
         """
         Perform a GET request and get the contents of the JSON response.

marathon_acme/tests/helpers.py

@@ -1,12 +1,39 @@
 from treq.client import HTTPClient
 from twisted.internet.defer import fail
+from uritools import urisplit
 
 
 class FailingAgent(object):
+    def __init__(self, error=RuntimeError()):
+        self.error = error
+
     """ A twisted.web.iweb.IAgent that does nothing but fail. """
     def request(self, method, uri, headers=None, bodyProducer=None):
-        return fail(RuntimeError())
+        return fail(self.error)
 
 
 """ A Treq client that will fail with a RuntimeError for any request made. """
 failing_client = HTTPClient(FailingAgent())
+
+
+class PerLocationAgent(object):
+    """
+    A twisted.web.iweb.IAgent that delegates to other agents for specific URI
+    locations.
+    """
+    def __init__(self):
+        self.agents = {}
+
+    def add_agent(self, location, agent):
+        """
+        Add an agent for URIs with the specified location.
+        :param bytes location:
+            The URI authority/location (e.g. b'example.com:80')
+        :param agent: The twisted.web.iweb.IAgent to use for the location
+        """
+        self.agents[location] = agent
+
+    def request(self, method, uri, headers=None, bodyProducer=None):
+        agent = self.agents[urisplit(uri).authority]
+        return agent.request(
+            method, uri, headers=headers, bodyProducer=bodyProducer)

marathon_acme/tests/test_clients.py

@@ -22,7 +22,8 @@
     json_content, JsonClient, MarathonClient, MarathonLbClient,
     raise_for_status)
 from marathon_acme.server import write_request_json
-from marathon_acme.tests.helpers import failing_client
+from marathon_acme.tests.helpers import (
+    failing_client, FailingAgent, PerLocationAgent)
 from marathon_acme.tests.matchers import (
     HasHeader, HasRequestProperties, WithErrorTypeAndMessage)
 
@@ -127,10 +128,10 @@ def setUp(self):
         super(TestHTTPClientBase, self).setUp()
 
         self.requests = DeferredQueue()
-        fake_server = FakeHttpServer(self.handle_request)
+        self.fake_server = FakeHttpServer(self.handle_request)
 
-        inner_client = treq_HTTPClient(fake_server.get_agent())
-        self.client = self.get_client(inner_client)
+        fake_client = treq_HTTPClient(self.fake_server.get_agent())
+        self.client = self.get_client(fake_client)
 
         # Spin the reactor once at the end of each test to clean up any
         # cancelled deferreds
@@ -519,7 +520,82 @@ def test_json_data_and_data_not_allowed(self):
 
 class TestMarathonClient(TestHTTPClientBase):
     def get_client(self, client):
-        return MarathonClient('http://localhost:8080', client=client)
+        return MarathonClient(['http://localhost:8080'], client=client)
+
+    def uri(self, path, index=0):
+        return ''.join((self.client.endpoints[index], path))
+
+    @inlineCallbacks
+    def test_request_success(self):
+        """
+        When we make a request and there are multiple Marathon endpoints
+        specified, the first endpoint is used.
+        """
+        agent = PerLocationAgent()
+        agent.add_agent(b'localhost:8080', self.fake_server.get_agent())
+        agent.add_agent(b'localhost:9090', FailingAgent())
+        client = MarathonClient(
+            ['http://localhost:8080', 'http://localhost:9090'],
+            client=treq_HTTPClient(agent))
+
+        d = self.cleanup_d(client.request('GET', path='/my-path'))
+
+        request = yield self.requests.get()
+        self.assertThat(request, HasRequestProperties(
+            method='GET', url='http://localhost:8080/my-path'))
+
+        request.setResponseCode(200)
+        request.finish()
+
+        yield d
+
+    @inlineCallbacks
+    def test_request_fallback(self):
+        """
+        When we make a request and there are multiple Marathon endpoints
+        specified, and an endpoint fails, the next endpoint is used.
+        """
+        agent = PerLocationAgent()
+        agent.add_agent(b'localhost:8080', FailingAgent())
+        agent.add_agent(b'localhost:9090', self.fake_server.get_agent())
+        client = MarathonClient(
+            ['http://localhost:8080', 'http://localhost:9090'],
+            client=treq_HTTPClient(agent))
+
+        d = self.cleanup_d(client.request('GET', path='/my-path'))
+
+        request = yield self.requests.get()
+        self.assertThat(request, HasRequestProperties(
+            method='GET', url='http://localhost:9090/my-path'))
+
+        request.setResponseCode(200)
+        request.finish()
+
+        yield d
+
+        flush_logged_errors(RuntimeError)
+
+    @inlineCallbacks
+    def test_request_fallback_all_failed(self):
+        """
+        When we make a request and there are multiple Marathon endpoints
+        specified, and all the endpoints fail, the last failure should be
+        returned.
+        """
+        agent = PerLocationAgent()
+        agent.add_agent(b'localhost:8080', FailingAgent(RuntimeError('8080')))
+        agent.add_agent(b'localhost:9090', FailingAgent(RuntimeError('9090')))
+        client = MarathonClient(
+            ['http://localhost:8080', 'http://localhost:9090'],
+            client=treq_HTTPClient(agent))
+
+        d = self.cleanup_d(client.request('GET', path='/my-path'))
+
+        yield wait0()
+        self.assertThat(d, failed(WithErrorTypeAndMessage(
+            RuntimeError, '9090')))
+
+        flush_logged_errors(RuntimeError)
 
     @inlineCallbacks
     def test_get_json_field(self):

marathon_acme/tests/test_service.py

@@ -92,7 +92,7 @@ def setup_method(self):
         self.fake_marathon = FakeMarathon()
         self.fake_marathon_api = FakeMarathonAPI(self.fake_marathon)
         marathon_client = MarathonClient(
-            'http://localhost:8080', client=self.fake_marathon_api.client)
+            ['http://localhost:8080'], client=self.fake_marathon_api.client)
 
         self.cert_store = MemoryStore()
 
@@ -154,7 +154,7 @@ def test_listen_events_attach_only_first(self):
 
         # Some other client attaches
         other_client = MarathonClient(
-            'http://localhost:8080', client=self.fake_marathon_api.client)
+            ['http://localhost:8080'], client=self.fake_marathon_api.client)
         other_client_events = []
         other_client.get_events(
             {'event_stream_attached': other_client_events.append})

marathon_acme/tests/test_test_helpers.py

@@ -0,0 +1,81 @@
+import pytest
+from testtools import ExpectedException
+from testtools.assertions import assert_that
+from testtools.matchers import (
+    Equals, Is, IsInstance, MatchesAll, MatchesDict, MatchesListwise,
+    MatchesPredicate, MatchesStructure)
+from testtools.twistedsupport import failed, succeeded
+from twisted.internet.defer import succeed
+
+from marathon_acme.tests.helpers import FailingAgent, PerLocationAgent
+
+
+class DummyAgent(object):
+    def request(self, *args, **kwargs):
+        return succeed((args, kwargs))
+
+
+class TestPerLocationAgent(object):
+    @pytest.fixture
+    def agent(self):
+        return PerLocationAgent()
+
+    def test_keyerror_if_location_unset(self, agent):
+        """
+        When a request is made using the agent and no delegate agent has been
+        added for the URI location/authority, a KeyError is expected.
+        """
+        with ExpectedException(KeyError, r"b?'foo:8080'"):
+            agent.request(b'GET', b'http://foo:8080')
+
+    def test_delegates_to_agent_for_location(self, agent):
+        """
+        When a request is made using the agent, the added agents are delegated
+        to based on the URI location/authority.
+        """
+        agent.add_agent(b'foo:8080', DummyAgent())
+        agent.add_agent(b'bar:8080', FailingAgent(RuntimeError('bar')))
+        agent.add_agent(b'foo:9090', FailingAgent(RuntimeError('9090')))
+
+        d = agent.request(b'GET', b'http://foo:8080')
+        assert_that(d, succeeded(MatchesListwise([
+            MatchesListwise([Equals(b'GET'), Equals(b'http://foo:8080')]),
+            MatchesDict({'headers': Is(None), 'bodyProducer': Is(None)})
+        ])))
+
+        # Scheme doesn't matter
+        d = agent.request(b'GET', b'https://foo:8080')
+        assert_that(d, succeeded(MatchesListwise([
+            MatchesListwise([Equals(b'GET'), Equals(b'https://foo:8080')]),
+            MatchesDict({'headers': Is(None), 'bodyProducer': Is(None)})
+        ])))
+
+        # Path doesn't matter
+        d = agent.request(b'GET', b'http://foo:8080/bar/baz')
+        assert_that(d, succeeded(MatchesListwise([
+            MatchesListwise([
+                Equals(b'GET'), Equals(b'http://foo:8080/bar/baz')]),
+            MatchesDict({'headers': Is(None), 'bodyProducer': Is(None)})
+        ])))
+
+        # Hostname *does* matter
+        d = agent.request(b'GET', b'http://bar:8080')
+        assert_that(d, failed(MatchesStructure(value=MatchesAll(
+            IsInstance(RuntimeError),
+            MatchesPredicate(str, Equals('bar'))
+        ))))
+
+        # Port *does* matter
+        d = agent.request(b'GET', b'http://foo:9090')
+        assert_that(d, failed(MatchesStructure(value=MatchesAll(
+            IsInstance(RuntimeError),
+            MatchesPredicate(str, Equals('9090'))
+        ))))
+
+        # Other args passed through
+        d = agent.request(b'GET', b'http://foo:8080', 'bar', 'baz')
+        assert_that(d, succeeded(MatchesListwise([
+            MatchesListwise([Equals(b'GET'), Equals(b'http://foo:8080')]),
+            MatchesDict(
+                {'headers': Equals('bar'), 'bodyProducer': Equals('baz')})
+        ])))