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.

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)

t/unit/backends/test_consul.py

@@ -22,8 +22,8 @@ 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_index_bytes_key(self):