redis: Support fallback servers

Sentinel clients can use fallback sentinel services so that clients can
obtain available instances even if some of the sentinel services are
unavailable.

Change-Id: Ibdf8993f7ca912bd5acf75ffbc89cf928b607b99

releasenotes/notes/sentinel-fallbacks-6fe2ab0d68959cdf.yaml

@@ -0,0 +1,5 @@
+---
+features:
+  - |
+    The redis driver now supports ``sentinel_fallbacks`` option. This allows
+    using additional sentinel servers as fallbacks.

taskflow/jobs/backends/impl_redis.py

@@ -17,6 +17,7 @@
 import contextlib
 import datetime
 import functools
+import re
 import string
 import threading
 import time
@@ -558,6 +559,18 @@ return cmsgpack.pack(result)
                                                           ut-were-afraid-to-ask
     """
 
+    @classmethod
+    def _parse_sentinel(cls, sentinel):
+        # IPv6 (eg. [::1]:6379 )
+        match = re.search(r'\[(\S+)\]:(\d+)', sentinel)
+        if match:
+            return (match[1], int(match[2]))
+        # IPv4 or hostname (eg. 127.0.0.1:6379 or localhost:6379)
+        match = re.search(r'(\S+):(\d+)', sentinel)
+        if match:
+            return (match[1], int(match[2]))
+        raise ValueError('Malformed sentinel server format')
+
     @classmethod
     def _make_client(cls, conf):
         client_conf = {}
@@ -569,6 +582,8 @@ return cmsgpack.pack(result)
                     client_conf[key] = conf[key]
         if conf.get('sentinel') is not None:
             sentinels = [(client_conf.pop('host'), client_conf.pop('port'))]
+            for fallback in conf.get('sentinel_fallbacks', []):
+                sentinels.append(cls._parse_sentinel(fallback))
             sentinel_kwargs = conf.get('sentinel_kwargs', {})
             for _opt in ('username', 'password', 'socket_timeout'):
                 if conf.get(_opt) is not None:

taskflow/tests/unit/jobs/test_redis_job.py

@@ -146,6 +146,32 @@ class RedisJobboardTest(test.TestCase, base.BoardTestMixin):
                 **test_conf)
             mock_sentinel().master_for.assert_called_once_with('mymaster')
 
+    def test__make_client_sentinel_fallbacks(self):
+        conf = {'host': '127.0.0.1',
+                'port': 26379,
+                'username': 'default',
+                'password': 'secret',
+                'namespace': 'test',
+                'sentinel': 'mymaster',
+                'sentinel_fallbacks': [
+                    '[::1]:26379', '127.0.0.2:26379', 'localhost:26379'
+                ]}
+        with mock.patch('redis.sentinel.Sentinel') as mock_sentinel:
+            impl_redis.RedisJobBoard('test-board', conf)
+            test_conf = {
+                'username': 'default',
+                'password': 'secret',
+            }
+            mock_sentinel.assert_called_once_with(
+                [('127.0.0.1', 26379), ('::1', 26379),
+                 ('127.0.0.2', 26379), ('localhost', 26379)],
+                sentinel_kwargs={
+                    'username': 'default',
+                    'password': 'secret'
+                },
+                **test_conf)
+            mock_sentinel().master_for.assert_called_once_with('mymaster')
+
     def test__make_client_sentinel_ssl(self):
         conf = {'host': '127.0.0.1',
                 'port': 26379,