Temporary hack. Redis Sentinel support for Celery.

celery_sentinel.py

"""
This module adds Redis Sentinel transport support to Celery.
Current version of celery doesn't support Redis sentinel client, which is must have for automatic failover.

To use it::

    import register_celery_alias
    register_celery_alias("redis-sentinel")

    celery = Celery(..., broker="redis-sentinel://...", backend="redis-sentinel://...")
"""
from celery.backends import BACKEND_ALIASES
from kombu.transport import TRANSPORT_ALIASES
from celery.backends.redis import RedisBackend
from kombu.utils import cached_property
from kombu.transport.redis import Transport, Channel
from redis import Redis
from redis.sentinel import Sentinel


class RedisSentinelBackend(RedisBackend):

    def __init__(self, sentinels=None, sentinel_timeout=None, socket_timeout=None,
                 min_other_sentinels=0, service_name=None, **kwargs):
        super(RedisSentinelBackend, self).__init__(**kwargs)
        conf = self.app.conf

        def _get(key):
            try:
                return conf['CELERY_REDIS_SENTINEL_%s' % key]
            except KeyError:
                pass

        self.sentinels = sentinels or _get("SENTINELS")
        self.sentinel_timeout = sentinel_timeout or _get("SENTINEL_TIMEOUT")
        self.socket_timeout = socket_timeout or _get("SOCKET_TIMEOUT")
        self.min_other_sentinels = min_other_sentinels or _get("MIN_OTHER_SENTINELS")
        self.service_name = service_name or _get("SERVICE_NAME")

    @cached_property
    def client(self):
        sentinel = Sentinel(self.sentinels, min_other_sentinels=self.min_other_sentinels,
                            password=self.password, sentinel_kwargs={"socket_timeout": self.sentinel_timeout})
        return sentinel.master_for(self.service_name, redis_class=Redis, socket_timeout=self.socket_timeout)


class SentinelChannel(Channel):

    from_transport_options = Channel.from_transport_options + (
        "service_name",
        "sentinels",
        "password",
        "min_other_sentinels",
        "sentinel_timeout",
    )

    #noinspection PyUnresolvedReferences
    @cached_property
    def _sentinel_managed_pool(self):

        sentinel = Sentinel(
            self.sentinels,
            min_other_sentinels=getattr(self, "min_other_sentinels", 0),
            password=getattr(self, "password", None),
            sentinel_kwargs={"socket_timeout": getattr(self, "sentinel_timeout", None)},
        )
        return sentinel.master_for(self.service_name, self.Client,
                                   socket_timeout=self.socket_timeout).connection_pool

    def _get_pool(self):
        return self._sentinel_managed_pool


class RedisSentinelTransport(Transport):
    Channel = SentinelChannel


def register_celery_alias(alias="redis-sentinel"):
    BACKEND_ALIASES[alias] = "utils.celery_sentinel.RedisSentinelBackend"
    TRANSPORT_ALIASES[alias] = "utils.celery_sentinel.RedisSentinelTransport"

thanks for this awesome hack that saved me tons of work!

Awesome 😃

@Surgo , @reclosedev : We have tried this hack for supporting redis-sentinel in celery ,but it is not working . celery result backend giving error as it is writing on the redis slave . by this hack celery is not able to make client connection through client function overridden in this hack to redis master node .Did any one have tried this before ?? Could you please help on this.

we just open-sourced from @dealertrack a library which adds sentinel support to celery which you can check out at http://celery-redis-sentinel.readthedocs.org/en/latest/. It adds support to both broker and results backend. Any feedback will be appreciated.