redis: Support Sentinel with SSL

Use the SentinelManagedSSLConnection when SSL is enabled for the
transport. The redis-py project doesn't have a connection class for
SSL+Sentinel yet. So, create a class in redis.py to add that
functionality.

celery/backends/redis.py

@@ -185,6 +185,7 @@ class RedisBackend(BaseKeyValueStoreBackend, AsyncBackendMixin):
 
     #: :pypi:`redis` client module.
     redis = redis
+    connection_class_ssl = redis.SSLConnection if redis else None
 
     #: Maximum number of connections in the pool.
     max_connections = None
@@ -236,7 +237,7 @@ def __init__(self, host=None, port=None, db=None, password=None,
         ssl = _get('redis_backend_use_ssl')
         if ssl:
             self.connparams.update(ssl)
-            self.connparams['connection_class'] = redis.SSLConnection
+            self.connparams['connection_class'] = self.connection_class_ssl
 
         if url:
             self.connparams = self._params_from_url(url, self.connparams)
@@ -245,7 +246,7 @@ def __init__(self, host=None, port=None, db=None, password=None,
         # redis_backend_use_ssl dict, check ssl_cert_reqs is valid. If set
         # via query string ssl_cert_reqs will be a string so convert it here
         if ('connection_class' in self.connparams and
-                self.connparams['connection_class'] is redis.SSLConnection):
+                issubclass(self.connparams['connection_class'], redis.SSLConnection)):
             ssl_cert_reqs_missing = 'MISSING'
             ssl_string_to_constant = {'CERT_REQUIRED': CERT_REQUIRED,
                                       'CERT_OPTIONAL': CERT_OPTIONAL,
@@ -535,10 +536,25 @@ def __reduce__(self, args=(), kwargs=None):
         )
 
 
+if getattr(redis, "sentinel", None):
+    class SentinelManagedSSLConnection(
+            redis.sentinel.SentinelManagedConnection,
+            redis.SSLConnection):
+        """Connect to a Redis server using Sentinel + TLS.
+
+        Use Sentinel to identify which Redis server is the current master
+        to connect to and when connecting to the Master server, use an
+        SSL Connection.
+        """
+
+        pass
+
+
 class SentinelBackend(RedisBackend):
     """Redis sentinel task result store."""
 
     sentinel = getattr(redis, "sentinel", None)
+    connection_class_ssl = SentinelManagedSSLConnection if sentinel else None
 
     def __init__(self, *args, **kwargs):
         if self.sentinel is None:

t/unit/backends/test_redis.py

@@ -1126,3 +1126,34 @@ def test_get_pool(self):
         )
         pool = x._get_pool(**x.connparams)
         assert pool
+
+    def test_backend_ssl(self):
+        pytest.importorskip('redis')
+
+        from celery.backends.redis import SentinelBackend
+        self.app.conf.redis_backend_use_ssl = {
+            'ssl_cert_reqs': "CERT_REQUIRED",
+            'ssl_ca_certs': '/path/to/ca.crt',
+            'ssl_certfile': '/path/to/client.crt',
+            'ssl_keyfile': '/path/to/client.key',
+        }
+        self.app.conf.redis_socket_timeout = 30.0
+        self.app.conf.redis_socket_connect_timeout = 100.0
+        x = SentinelBackend(
+            'sentinel://:bosco@vandelay.com:123//1', app=self.app,
+        )
+        assert x.connparams
+        assert len(x.connparams['hosts']) == 1
+        assert x.connparams['hosts'][0]['host'] == 'vandelay.com'
+        assert x.connparams['hosts'][0]['db'] == 1
+        assert x.connparams['hosts'][0]['port'] == 123
+        assert x.connparams['hosts'][0]['password'] == 'bosco'
+        assert x.connparams['socket_timeout'] == 30.0
+        assert x.connparams['socket_connect_timeout'] == 100.0
+        assert x.connparams['ssl_cert_reqs'] == ssl.CERT_REQUIRED
+        assert x.connparams['ssl_ca_certs'] == '/path/to/ca.crt'
+        assert x.connparams['ssl_certfile'] == '/path/to/client.crt'
+        assert x.connparams['ssl_keyfile'] == '/path/to/client.key'
+
+        from celery.backends.redis import SentinelManagedSSLConnection
+        assert x.connparams['connection_class'] is SentinelManagedSSLConnection