The Consul backend must correctly associate requests and responses (celery#6823)

* As per celery#5605, the Consul backend does not cleanly associate responses
from Consul with the outbound Celery request that caused it. This leaves
it prone to mistaking the (final) response from an operation N as the
response to an (early) part of operation N + 1.

This changes fix that by using a separate connection for each request.
That of course has the downside of (a) being relatively expensive and (b)
increasing the rate of connection requests into Consul:

- The former is annoying, but at least the backend works reliably.

- The latter can cause Consul to reject excessive connection attempt, but
  if it does, at least it returns a clear indication of this (IIRC, it
  responds with an HTTP 429"too many connections" indication).

  Additionally, this issue can be ameliorated by enabling retries in the
  python-consul2 (which I believe should be turned on regards less to handle
  transient network issues). This is fixed by the PR in
  https:/github.com/poppyred/python-consul2/pull/31.

Note that we have never seen (b) outside a test specifically trying to hammer
the system, but we see (a) all the time in our normal system tests.

To opt-out from the new behaviour add a parameter "one_client=1" to the
connection URL.

* Increase code coverage.

* Rewrite Consul backend documentation, and describe the options now
available.

Author: ShaheedHaque

celery/backends/consul.py

@@ -31,7 +31,6 @@ class ConsulBackend(KeyValueStoreBackend):
 
     supports_autoexpire = True
 
-    client = None
     consistency = 'consistent'
     path = None
 
@@ -40,15 +39,33 @@ def __init__(self, *args, **kwargs):
 
         if self.consul is None:
             raise ImproperlyConfigured(CONSUL_MISSING)
-
+        #
+        # By default, for correctness, we use a client connection per
+        # operation. If set, self.one_client will be used for all operations.
+        # This provides for the original behaviour to be selected, and is
+        # also convenient for mocking in the unit tests.
+        #
+        self.one_client = None
         self._init_from_params(**parse_url(self.url))
 
     def _init_from_params(self, hostname, port, virtual_host, **params):
         logger.debug('Setting on Consul client to connect to %s:%d',
                      hostname, port)
         self.path = virtual_host
-        self.client = consul.Consul(host=hostname, port=port,
-                                    consistency=self.consistency)
+        self.hostname = hostname
+        self.port = port
+        #
+        # Optionally, allow a single client connection to be used to reduce
+        # the connection load on Consul by adding a "one_client=1" parameter
+        # to the URL.
+        #
+        if params.get('one_client', None):
+            self.one_client = self.client()
+
+    def client(self):
+        return self.one_client or consul.Consul(host=self.hostname,
+                                                port=self.port,
+                                                consistency=self.consistency)
 
     def _key_to_consul_key(self, key):
         key = bytes_to_str(key)
@@ -58,7 +75,7 @@ def get(self, key):
         key = self._key_to_consul_key(key)
         logger.debug('Trying to fetch key %s from Consul', key)
         try:
-            _, data = self.client.kv.get(key)
+            _, data = self.client().kv.get(key)
             return data['Value']
         except TypeError:
             pass
@@ -84,17 +101,16 @@ def set(self, key, value):
 
         logger.debug('Trying to create Consul session %s with TTL %d',
                      session_name, self.expires)
-        session_id = self.client.session.create(name=session_name,
-                                                behavior='delete',
-                                                ttl=self.expires)
+        client = self.client()
+        session_id = client.session.create(name=session_name,
+                                           behavior='delete',
+                                           ttl=self.expires)
         logger.debug('Created Consul session %s', session_id)
 
         logger.debug('Writing key %s to Consul', key)
-        return self.client.kv.put(key=key,
-                                  value=value,
-                                  acquire=session_id)
+        return client.kv.put(key=key, value=value, acquire=session_id)
 
     def delete(self, key):
         key = self._key_to_consul_key(key)
         logger.debug('Removing key %s from Consul', key)
-        return self.client.kv.delete(key)
+        return self.client().kv.delete(key)

docs/userguide/configuration.rst

@@ -2016,14 +2016,52 @@ without any further configuration. For larger clusters you could use NFS,
 Consul K/V store backend settings
 ---------------------------------
 
-The Consul backend can be configured using a URL, for example:
+.. note::
+
+    The Consul backend requires the :pypi:`python-consul2` library:
+
+    To install this package use :command:`pip`:
+
+    .. code-block:: console
+
+        $ pip install python-consul2
+
+The Consul backend can be configured using a URL, for example::
 
     CELERY_RESULT_BACKEND = 'consul://localhost:8500/'
 
-The backend will storage results in the K/V store of Consul
-as individual keys.
+or::
+
+    result_backend = 'consul://localhost:8500/'
+
+The backend will store results in the K/V store of Consul
+as individual keys. The backend supports auto expire of results using TTLs in
+Consul. The full syntax of the URL is::
+
+    consul://host:port[?one_client=1]
+
+The URL is formed out of the following parts:
+
+* ``host``
+
+    Host name of the Consul server.
+
+* ``port``
+
+    The port the Consul server is listening to.
+
+* ``one_client``
+
+    By default, for correctness, the backend uses a separate client connection
+    per operation. In cases of extreme load, the rate of creation of new
+    connections can cause HTTP 429 "too many connections" error responses from
+    the Consul server when under load. The recommended way to handle this is to
+    enable retries in ``python-consul2`` using the patch at
+    https://github.com/poppyred/python-consul2/pull/31.
 
-The backend supports auto expire of results using TTLs in Consul.
+    Alternatively, if ``one_client`` is set, a single client connection will be
+    used for all operations instead. This should eliminate the HTTP 429 errors,
+    but the storage of results in the backend can become unreliable.
 
 .. _conf-messaging:
 

t/unit/backends/test_consul.py

@@ -22,10 +22,21 @@ def test_consul_consistency(self):
     def test_get(self):
         index = 100
         data = {'Key': 'test-consul-1', 'Value': 'mypayload'}
-        self.backend.client = Mock(name='c.client')
-        self.backend.client.kv.get.return_value = (index, data)
+        self.backend.one_client = Mock(name='c.client')
+        self.backend.one_client.kv.get.return_value = (index, data)
         assert self.backend.get(data['Key']) == 'mypayload'
 
+    def test_set(self):
+        self.backend.one_client = Mock(name='c.client')
+        self.backend.one_client.session.create.return_value = 'c8dfa770-4ea3-2ee9-d141-98cf0bfe9c59'
+        self.backend.one_client.kv.put.return_value = True
+        assert self.backend.set('Key', 'Value') is True
+
+    def test_delete(self):
+        self.backend.one_client = Mock(name='c.client')
+        self.backend.one_client.kv.delete.return_value = True
+        assert self.backend.delete('Key') is True
+
     def test_index_bytes_key(self):
         key = 'test-consul-2'
         assert self.backend._key_to_consul_key(key) == key