#97 - Add support for redis sentinel

celery.js

@@ -1,15 +1,15 @@
 var url = require('url'),
     util = require('util'),
     amqp = require('amqp'),
-    redis = require('redis'),
+    Redis = require('ioredis'),
     events = require('events'),
     uuid = require('uuid');
 
 var createMessage = require('./protocol').createMessage;
 
 var debug = process.env.NODE_CELERY_DEBUG === '1' ? console.info : function() {};
 
-var supportedProtocols = ['amqp', 'amqps', 'redis'];
+var supportedProtocols = ['amqp', 'amqps', 'redis', 'sentinel'];
 function getProtocol(kind, options) {
     const protocol = url.parse(options.url).protocol.slice(0, -1);
     if (protocol === 'amqps') {
@@ -68,7 +68,7 @@ function Configuration(options) {
 
 function RedisBroker(conf) {
     var self = this;
-    self.redis = redis.createClient(conf.BROKER_OPTIONS);
+    self.redis = new Redis(conf.BROKER_OPTIONS);
 
     self.end = function() {
         self.redis.end(true);
@@ -118,8 +118,7 @@ util.inherits(RedisBroker, events.EventEmitter);
 
 function RedisBackend(conf) {
     var self = this;
-    self.redis = redis.createClient(conf.RESULT_BACKEND_OPTIONS);
-
+    self.redis = new Redis(conf.RESULT_BACKEND_OPTIONS);
     var backend_ex = self.redis.duplicate();
 
     self.redis.on('error', function(err) {
@@ -179,7 +178,7 @@ function Client(conf) {
     self.conf = new Configuration(conf);
 
     // backend
-    if (self.conf.backend_type === 'redis') {
+    if (self.conf.backend_type === 'redis' || self.conf.backend_type === 'sentinel') {
         self.backend = new RedisBackend(self.conf);
         self.backend.on('message', function(msg) {
             self.emit('message', msg);
@@ -198,7 +197,7 @@ function Client(conf) {
     self.backend.on('ready', function() {
         debug('Connecting to broker...');
 
-        if (self.conf.broker_type === 'redis') {
+        if (self.conf.broker_type === 'redis' || self.conf.broker_type === 'sentinel') {
             self.broker = new RedisBroker(self.conf);
         } else if (self.conf.broker_type === 'amqp') {
             self.broker = amqp.createConnection(self.conf.BROKER_OPTIONS, {
@@ -252,7 +251,6 @@ Client.prototype.call = function(name /*[args], [kwargs], [options], [callback]*
 
     var task = this.createTask(name),
         result = task.call(args, kwargs, options);
-
     if (callback && result) {
         debug('Subscribing to result...');
         result.on('ready', callback);
@@ -275,7 +273,7 @@ function Task(client, name, options, exchange) {
 
         var result = new Result(id, self.client);
 
-        if (client.conf.backend_type === 'redis') {
+        if (client.conf.backend_type === 'redis' || client.conf.backend_type === 'sentinel') {
             client.backend.results[result.taskid] = result;
         }
 

dockers/.env

@@ -0,0 +1 @@
+EXTERNAL_HOST=10.10.2.51

dockers/celery4.2/Dockerfile

@@ -0,0 +1,23 @@
+FROM python:3.5-slim
+
+RUN groupadd user && useradd --create-home --home-dir /home/user -g user user
+WORKDIR /home/user
+
+RUN pip install redis
+
+ENV CELERY_VERSION 4.1
+
+RUN pip install celery=="$CELERY_VERSION"
+
+RUN { \
+	echo 'import os'; \
+	echo "broker_url = os.environ.get('BROKER_URL', 'sentinel://sentinel:26379/')"; \
+  echo "broker_transport_options = { 'master_name': os.environ.get('MASTER_NAME', 'sentinelTest') }"; \
+	echo "backend_url = os.environ.get('BACKEND_URL', 'redis://redis/0')"; \
+} > celeryconfig.py
+
+# --link some-rabbit:rabbit "just works"
+# ENV BROKER_URL amqp://guest@rabbit
+
+USER user
+CMD ["celery", "worker"]

dockers/docker-compose.yml

@@ -0,0 +1,66 @@
+version: "3"
+
+services:
+  redis:
+    image: redis
+    ports:
+      - "6379:6379"
+
+  rabbit:
+    image: rabbitmq
+    ports:
+      - "5672:5672"
+
+  celery_amqp:
+    image: celery
+    command: celery worker -A tasks --loglevel=INFO
+    volumes:
+      - "./tasks.py:/home/user/tasks.py"
+    depends_on:
+      - rabbit
+
+  celery_redis:
+    image: celery
+    command: celery worker -A tasks --loglevel=INFO
+    volumes:
+      - "./tasks.py:/home/user/tasks.py"
+    environment:
+      - CELERY_BROKER_URL=redis://redis/0
+      - CELERY_BACKEND_URL=redis://redis/0
+    depends_on:
+      - redis
+
+  redis_master:
+    image: 'redis'
+    command: redis-server --port 6380
+    ports:
+      - '6380:6380'
+
+  redis_slave:
+    image: 'redis'
+    command: redis-server --port 6381 --slaveof "${EXTERNAL_HOST}" 6380 --slave-announce-ip "${EXTERNAL_HOST}"
+    ports:
+      - '6381:6381'
+
+  sentinel:
+    build: ./sentinel
+    ports:
+      - '26379:26379'
+    environment:
+      - SENTINEL_NAME=sentinelTest
+      - HOST_IP="${EXTERNAL_HOST}"
+    depends_on:
+      - redis_master
+      - redis_slave
+
+  celery_sentinel:
+    build: ./celery4.2
+    command: celery worker -A tasks --loglevel=INFO
+    volumes:
+      - "./sentinel_tasks.py:/home/user/tasks.py"
+    environment:
+      - BROKER_URL=sentinel://sentinel:26379
+      - MASTER_NAME=sentinelTest
+      - BACKEND_URL=redis://redis_master:6380/0
+    depends_on:
+      - redis

dockers/down.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+set -ev
+
+pushd `dirname $0`
+docker-compose down
+docker-compose ps
+popd

dockers/sentinel/Dockerfile

@@ -0,0 +1,11 @@
+FROM redis
+
+EXPOSE 26379
+COPY sentinel.conf /etc/redis/sentinel.conf
+RUN chown redis:redis /etc/redis/sentinel.conf
+ENV SENTINEL_QUORUM 2
+ENV SENTINEL_NAME sentinelTest
+ENV SENTINEL_DOWN_AFTER 30000
+ENV SENTINEL_FAILOVER 180000
+COPY sentinel-entrypoint.sh /usr/local/bin/
+ENTRYPOINT ["sentinel-entrypoint.sh"]

dockers/sentinel/sentinel-entrypoint.sh

@@ -0,0 +1,9 @@
+#!/bin/sh
+
+sed -i "s/\$SENTINEL_QUORUM/$SENTINEL_QUORUM/g" /etc/redis/sentinel.conf
+sed -i "s/\$SENTINEL_DOWN_AFTER/$SENTINEL_DOWN_AFTER/g" /etc/redis/sentinel.conf
+sed -i "s/\$SENTINEL_FAILOVER/$SENTINEL_FAILOVER/g" /etc/redis/sentinel.conf
+sed -i "s/\$SENTINEL_NAME/$SENTINEL_NAME/g" /etc/redis/sentinel.conf
+sed -i "s/\$HOST_IP/$HOST_IP/g" /etc/redis/sentinel.conf
+
+exec docker-entrypoint.sh redis-server /etc/redis/sentinel.conf --sentinel

dockers/sentinel/sentinel.conf

@@ -0,0 +1,11 @@
+port 26379
+
+dir /tmp
+
+sentinel monitor $SENTINEL_NAME $HOST_IP 6380 $SENTINEL_QUORUM
+
+sentinel down-after-milliseconds $SENTINEL_NAME $SENTINEL_DOWN_AFTER
+
+sentinel parallel-syncs $SENTINEL_NAME 1
+
+sentinel failover-timeout $SENTINEL_NAME $SENTINEL_FAILOVER

dockers/sentinel_tasks.py

@@ -0,0 +1,48 @@
+import os
+import logging
+from celery import Celery
+
+celery = Celery('tasks')
+
+celery.conf.broker_url = os.environ.get('BROKER_URL')
+celery.conf.broker_transport_options = {
+    'master_name': os.environ.get('MASTER_NAME')
+}
+celery.conf.result_backend = os.environ.get('BACKEND_URL')
+
+celery.conf.update(
+    result_serializer='json',
+    enable_utc=True
+)
+
+
+@celery.task
+def add(x, y):
+    return x + y
+
+
+@celery.task
+def sleep(x):
+    time.sleep(x)
+    return x
+
+
+@celery.task
+def time():
+    import time
+    return time.time()
+
+
+@celery.task
+def error(msg):
+    raise Exception(msg)
+
+
+@celery.task
+def echo(msg):
+    return msg
+
+
+@celery.task
+def send_email(to='me@example.com', title='hi'):
+    logging.info("Sending email to '%s' with title '%s'" % (to, title))

dockers/up.sh

@@ -0,0 +1,16 @@
+#!/bin/bash
+set -ev
+
+pushd `dirname $0`
+
+# Get external IP of localhost
+IP=`ifconfig | grep -Eo 'inet (addr:)?([0-9]*\.){3}[0-9]*' | grep -Eo '([0-9]*\.){3}[0-9]*' | grep -v '127.0.0.1'`
+echo $IP
+
+echo "EXTERNAL_HOST=$IP" > .env
+
+# Start the services.
+docker-compose up -d --build
+docker-compose ps
+
+popd

package.json

@@ -4,11 +4,12 @@
   "description": "Celery client for Node",
   "author": "Mher Movsisyan <mher.movsisyan@gmail.com>",
   "scripts": {
-    "test": "mocha tests"
+    "test": "mocha tests --exit"
   },
   "dependencies": {
     "amqp": "*",
-    "redis": "*",
+    "ioredis": "^3.2.2",
+    "redis": "^2.8.0",
     "uuid": "^3.0.0"
   },
   "repository": {