Catch Redis server errors

gnocchi/incoming/redis.py

@@ -17,10 +17,13 @@
 import uuid
 
 import daiquiri
+from redis.exceptions import ConnectionError
 import six
+import tenacity
 
 from gnocchi.common import redis
 from gnocchi import incoming
+from gnocchi import utils
 
 
 LOG = daiquiri.getLogger(__name__)
@@ -93,6 +96,10 @@ def add_measures_batch(self, metrics_and_measures):
                 notified_sacks.add(sack_name)
         pipe.execute()
 
+    @tenacity.retry(
+        wait=utils.wait_exponential,
+        # Never retry except when explicitly asked by raising TryAgain
+        retry=tenacity.retry_if_exception_type(ConnectionError))
     def _build_report(self, details):
         report_vars = {'measures': 0, 'metric_details': {}}
 
@@ -106,20 +113,23 @@ def update_report(results, m_list):
         metrics = 0
         m_list = []
         pipe = self._client.pipeline()
-        for key in self._client.scan_iter(match=match, count=1000):
-            metrics += 1
-            pipe.llen(key)
-            if details:
-                m_list.append(key.split(redis.SEP)[1].decode("utf8"))
-            # group 100 commands/call
-            if metrics % 100 == 0:
+        try:
+            for key in self._client.scan_iter(match=match, count=1000):
+                metrics += 1
+                pipe.llen(key)
+                if details:
+                    m_list.append(key.split(redis.SEP)[1].decode("utf8"))
+                # group 100 commands/call
+                if metrics % 100 == 0:
+                    results = pipe.execute()
+                    update_report(results, m_list)
+                    m_list = []
+                    pipe = self._client.pipeline()
+            else:
                 results = pipe.execute()
                 update_report(results, m_list)
-                m_list = []
-                pipe = self._client.pipeline()
-        else:
-            results = pipe.execute()
-            update_report(results, m_list)
+        except ConnectionError:
+            LOG.debug("Redis Server closed connection. Retrying.")
         return (metrics, report_vars['measures'],
                 report_vars['metric_details'] if details else None)
 
@@ -176,18 +186,25 @@ def process_measures_for_sack(self, sack):
             pipe.ltrim(key, item_len + 1, -1)
         pipe.execute()
 
+    @tenacity.retry(
+        wait=utils.wait_exponential,
+        # Never retry except when explicitly asked by raising TryAgain
+        retry=tenacity.retry_if_exception_type(ConnectionError))
     def iter_on_sacks_to_process(self):
         self._client.config_set("notify-keyspace-events", "K$")
         p = self._client.pubsub()
         db = self._client.connection_pool.connection_kwargs['db']
         keyspace = b"__keyspace@" + str(db).encode() + b"__:"
         pattern = keyspace + self._get_sack_name("*").encode()
         p.psubscribe(pattern)
-        for message in p.listen():
-            if message['type'] == 'pmessage' and message['pattern'] == pattern:
-                # FIXME(jd) This is awful, we need a better way to extract this
-                # Format is defined by _get_sack_name: incoming128-17
-                yield self._make_sack(int(message['channel'].split(b"-")[-1]))
+        try:
+            for message in p.listen():
+                if message['type'] == 'pmessage' and message['pattern'] == pattern:
+                    # FIXME(jd) This is awful, we need a better way to extract this
+                    # Format is defined by _get_sack_name: incoming128-17
+                    yield self._make_sack(int(message['channel'].split(b"-")[-1]))
+        except ConnectionError:
+            LOG.debug("Redis Server closed connection. Retrying.")
 
     def finish_sack_processing(self, sack):
         # Delete the sack key which handles no data but is used to get a SET

setup.cfg

@@ -70,7 +70,7 @@ s3 =
     boto3
     botocore>=1.5
 redis =
-    redis>=2.10.0 # MIT
+    redis >= 3.2.0 # MIT
     hiredis
 swift =
     python-swiftclient>=3.1.0