diff --git a/.travis.yml b/.travis.yml index eff67cd..b7b557c 100644 --- a/.travis.yml +++ b/.travis.yml @@ -11,6 +11,19 @@ env: - DJANGO=1.8.4 - DJANGO=1.7.10 +sudo: false + +addons: + apt: + packages: + - tcl8.5 + +before_script: + - git clone https://github.com/antirez/disque.git disque_server + - "cd disque_server/src && make && PREFIX=../ make install && cd -" + - "./disque_server/bin/disque-server &" + - ./disque_server/bin/disque PING + install: - pip install -q django==$DJANGO - pip install -r requirements.txt diff --git a/MANIFEST.in b/MANIFEST.in index 7a75164..b3169f4 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -2,4 +2,5 @@ include LICENSE include README.rst include django_q/management/*.py include django_q/management/commands/*.py -include django_q/migrations/*.py \ No newline at end of file +include django_q/migrations/*.py +include django_q/brokers/*.py \ No newline at end of file diff --git a/README.rst b/README.rst index 9f8550b..0909c8d 100644 --- a/README.rst +++ b/README.rst @@ -20,13 +20,12 @@ Features - Django Admin integration - PaaS compatible with multiple instances - Multi cluster monitor -- Redis broker +- Redis broker and Disque broker - Python 2 and 3 Requirements ~~~~~~~~~~~~ -- `Redis-py `__ - `Django `__ > = 1.7 - `Django-picklefield `__ - `Arrow `__ @@ -34,6 +33,13 @@ Requirements Tested with: Python 2.7 & 3.4. Django 1.7.10 & 1.8.4 +Brokers +~~~~~~~ +- `Redis `__ +- `Disque `__ +- `Amazon SQS `__ (TBA) +- `IronMQ `__ (TBA) + Installation ~~~~~~~~~~~~ @@ -54,8 +60,7 @@ Installation $ python manage.py migrate -- Make sure you have a `Redis `__ server running - somewhere +- Choose a message `broker `__ , configure and install the appropriate client library. Read the full documentation at `https://django-q.readthedocs.org `__ diff --git a/django_q/__init__.py b/django_q/__init__.py index c0913fb..9fc91f1 100644 --- a/django_q/__init__.py +++ b/django_q/__init__.py @@ -9,6 +9,6 @@ from .models import Task, Schedule, Success, Failure from .cluster import Cluster from .status import Stat -VERSION = (0, 5, 3) +VERSION = (0, 6, 0) default_app_config = 'django_q.apps.DjangoQConfig' diff --git a/django_q/brokers/__init__.py b/django_q/brokers/__init__.py new file mode 100644 index 0000000..c247bb1 --- /dev/null +++ b/django_q/brokers/__init__.py @@ -0,0 +1,159 @@ +from django_q.conf import Conf +from django.core.cache import caches, InvalidCacheBackendError + + +class Broker(object): + def __init__(self, list_key=Conf.PREFIX): + self.connection = self.get_connection(list_key) + self.list_key = list_key + self.cache = self.get_cache() + + def enqueue(self, task): + """ + Puts a task onto the queue + :type task: str + :return: task id + """ + pass + + def dequeue(self): + """ + Gets a task from the queue + :return: tuple with task id and task message + """ + pass + + def queue_size(self): + """ + :return: the amount of tasks in the queue + """ + pass + + def delete_queue(self): + """ + Deletes the queue from the broker + """ + pass + + def purge_queue(self): + """ + Purges the queue of any tasks + """ + pass + + def delete(self, task_id): + """ + Deletes a task from the queue + :param task_id: the id of the task + """ + pass + + def acknowledge(self, task_id): + """ + Acknowledges completion of the task and removes it from the queue. + :param task_id: the id of the task + """ + pass + + def fail(self, task_id): + """ + Fails a task message + :param task_id: + :return: + """ + + def ping(self): + """ + Checks whether the broker connection is available + :rtype: bool + """ + pass + + def info(self): + """ + Shows the broker type + """ + pass + + def set_stat(self, key, value, timeout): + """ + Saves a cluster statistic to the cache provider + :type key: str + :type value: str + :type timeout: int + """ + if not self.cache: + return + key_list = self.cache.get(Conf.Q_STAT, []) + if key not in key_list: + key_list.append(key) + self.cache.set(Conf.Q_STAT, key_list) + return self.cache.set(key, value, timeout) + + def get_stat(self, key): + """ + Gets a cluster statistic from the cache provider + :type key: str + :return: a cluster Stat + """ + if not self.cache: + return + return self.cache.get(key) + + def get_stats(self, pattern): + """ + Returns a list of all cluster stats from the cache provider + :type pattern: str + :return: a list of Stats + """ + if not self.cache: + return + key_list = self.cache.get(Conf.Q_STAT) + if not key_list or len(key_list) == 0: + return [] + stats = [] + for key in key_list: + stat = self.cache.get(key) + if stat: + stats.append(stat) + else: + key_list.remove(key) + self.cache.set(Conf.Q_STAT, key_list) + return stats + + @staticmethod + def get_cache(): + """ + Gets the current cache provider + :return: a cache provider + """ + try: + return caches[Conf.CACHE] + except InvalidCacheBackendError: + return None + + @staticmethod + def get_connection(list_key=Conf.PREFIX): + """ + Gets a connection to the broker + :param list_key: Optional queue name + :return: a broker connection + """ + return 0 + + +def get_broker(list_key=Conf.PREFIX): + """ + Gets the configured broker type + :param list_key: optional queue name + :type list_key: str + :return: + """ + # disque + if Conf.DISQUE_NODES: + from brokers import disque + return disque.Disque(list_key=list_key) + # default to redis + else: + from brokers import redis_broker + return redis_broker.Redis(list_key=list_key) diff --git a/django_q/brokers/disque.py b/django_q/brokers/disque.py new file mode 100644 index 0000000..7e43ccd --- /dev/null +++ b/django_q/brokers/disque.py @@ -0,0 +1,61 @@ +import random +import redis +from django_q.brokers import Broker +from django_q.conf import Conf + + +class Disque(Broker): + def enqueue(self, task): + retry = Conf.RETRY if Conf.RETRY > 0 else '{} REPLICATE 1'.format(Conf.RETRY) + return self.connection.execute_command( + 'ADDJOB {} {} 500 RETRY {}'.format(self.list_key, task, retry)).decode() + + def dequeue(self): + task = self.connection.execute_command('GETJOB TIMEOUT 1000 FROM {}'.format(self.list_key)) + if task: + return task[0][1].decode(), task[0][2].decode() + + def queue_size(self): + return self.connection.execute_command('QLEN {}'.format(self.list_key)) + + def acknowledge(self, task_id): + return self.connection.execute_command('ACKJOB {}'.format(task_id)) + + def ping(self): + return self.connection.execute_command('HELLO')[0] > 0 + + def delete(self, task_id): + return self.connection.execute_command('DELJOB {}'.format(task_id)) + + def fail(self, task_id): + return self.delete(task_id) + + def delete_queue(self): + jobs = self.connection.execute_command('JSCAN QUEUE {}'.format(self.list_key))[1] + if jobs: + job_ids = ' '.join(jid.decode() for jid in jobs) + self.connection.execute_command('DELJOB {}'.format(job_ids)) + return len(jobs) + + def info(self): + info = self.connection.info('server') + return 'Disque {}'.format(info['disque_version']) + + @staticmethod + def get_connection(list_key=Conf.PREFIX): + # randomize nodes + random.shuffle(Conf.DISQUE_NODES) + # find one that works + for node in Conf.DISQUE_NODES: + host, port = node.split(':') + kwargs = {'host': host, 'port': port} + if Conf.DISQUE_AUTH: + kwargs['password'] = Conf.DISQUE_AUTH + redis_client = redis.Redis(**kwargs) + redis_client.decode_responses = True + try: + redis_client.execute_command('HELLO') + return redis_client + except redis.exceptions.ConnectionError: + continue + raise redis.exceptions.ConnectionError('Could not connect to any Disque nodes') diff --git a/django_q/brokers/redis_broker.py b/django_q/brokers/redis_broker.py new file mode 100644 index 0000000..68289aa --- /dev/null +++ b/django_q/brokers/redis_broker.py @@ -0,0 +1,60 @@ +import redis +from django_q.brokers import Broker +from django_q.conf import Conf, logger + +try: + import django_redis +except ImportError: + django_redis = None + + +class Redis(Broker): + + def __init__(self, list_key=Conf.PREFIX): + super(Redis, self).__init__(list_key='django_q:{}:q'.format(list_key)) + + def enqueue(self, task): + return self.connection.rpush(self.list_key, task) + + def dequeue(self): + task = self.connection.blpop(self.list_key, 1) + if task: + return None, task[1] + + def queue_size(self): + return self.connection.llen(self.list_key) + + def delete_queue(self): + return self.connection.delete(self.list_key) + + def purge_queue(self): + return self.connection.ltrim(self.list_key, 1, 0) + + def ping(self): + try: + return self.connection.ping() + except redis.ConnectionError as e: + logger.error('Can not connect to Redis server.') + raise e + + def info(self): + info = self.connection.info('server') + return 'Redis {}'.format(info['redis_version']) + + def set_stat(self, key, value, timeout): + self.connection.set(key, value, timeout) + + def get_stat(self, key): + if self.connection.exists(key): + return self.connection.get(key) + + def get_stats(self, pattern): + keys = self.connection.keys(pattern=pattern) + if keys: + return self.connection.mget(keys) + + @staticmethod + def get_connection(list_key=Conf.PREFIX): + if django_redis and Conf.DJANGO_REDIS: + return django_redis.get_redis_connection(Conf.DJANGO_REDIS) + return redis.StrictRedis(**Conf.REDIS) diff --git a/django_q/cluster.py b/django_q/cluster.py index 1e03148..d3b5663 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -30,19 +30,20 @@ from django import db import signing import tasks -from django_q.conf import Conf, redis_client, logger, psutil, get_ppid +from django_q.conf import Conf, logger, psutil, get_ppid from django_q.models import Task, Success, Schedule -from django_q.status import Stat, Status, ping_redis +from django_q.status import Stat, Status +from django_q.brokers import get_broker class Cluster(object): - def __init__(self, list_key=Conf.Q_LIST): + def __init__(self, broker=None): + self.broker = broker or get_broker() self.sentinel = None self.stop_event = None self.start_event = None self.pid = current_process().pid self.host = socket.gethostname() - self.list_key = list_key self.timeout = Conf.TIMEOUT signal.signal(signal.SIGTERM, self.sig_handler) signal.signal(signal.SIGINT, self.sig_handler) @@ -58,7 +59,7 @@ class Cluster(object): self.stop_event = Event() self.start_event = Event() self.sentinel = Process(target=Sentinel, - args=(self.stop_event, self.start_event, self.list_key, self.timeout)) + args=(self.stop_event, self.start_event, self.broker, self.timeout)) self.sentinel.start() logger.info(_('Q Cluster-{} starting.').format(self.pid)) while not self.start_event.is_set(): @@ -105,15 +106,14 @@ class Cluster(object): class Sentinel(object): - def __init__(self, stop_event, start_event, list_key=Conf.Q_LIST, timeout=Conf.TIMEOUT, start=True): + def __init__(self, stop_event, start_event, broker=None, timeout=Conf.TIMEOUT, start=True): # Make sure we catch signals for the pool signal.signal(signal.SIGINT, signal.SIG_IGN) signal.signal(signal.SIGTERM, signal.SIG_DFL) self.pid = current_process().pid self.parent_pid = get_ppid() self.name = current_process().name - self.list_key = list_key - self.r = redis_client + self.broker = broker or get_broker() self.reincarnations = 0 self.tob = timezone.now() self.stop_event = stop_event @@ -130,7 +130,7 @@ class Sentinel(object): self.start() def start(self): - ping_redis(self.r) + self.broker.ping() self.spawn_cluster() self.guard() @@ -165,7 +165,7 @@ class Sentinel(object): return p def spawn_pusher(self): - return self.spawn_process(pusher, self.task_queue, self.event_out, self.list_key) + return self.spawn_process(pusher, self.task_queue, self.event_out, self.broker) def spawn_worker(self): self.spawn_process(worker, self.task_queue, self.result_queue, Value('f', -1), self.timeout) @@ -216,7 +216,7 @@ class Sentinel(object): self.start_event.set() Stat(self).save() logger.info(_('Q Cluster-{} running.').format(self.parent_pid)) - scheduler(list_key=self.list_key) + scheduler(broker=self.broker) counter = 0 cycle = 0.5 # guard loop sleep in seconds # Guard loop. Runs at least once @@ -240,7 +240,7 @@ class Sentinel(object): counter += cycle if counter == 30: counter = 0 - scheduler(list_key=self.list_key) + scheduler(broker=self.broker) # Save current status Stat(self).save() sleep(cycle) @@ -287,46 +287,54 @@ class Sentinel(object): Stat(self).save() -def pusher(task_queue, event, list_key=Conf.Q_LIST): +def pusher(task_queue, event, broker=None): """ Pulls tasks of the Redis List and puts them in the task queue :type task_queue: multiprocessing.Queue :type event: multiprocessing.Event - :type list_key: str """ + if not broker: + broker = get_broker() logger.info(_('{} pushing tasks at {}').format(current_process().name, current_process().pid)) - r = redis_client while True: try: - task = r.blpop(list_key, 1) + task = broker.dequeue() except Exception as e: logger.error(e) - # redis probably crashed. Let the sentinel handle it. + # broker probably crashed. Let the sentinel handle it. sleep(10) break if task: + ack_id = task[0] # unpack the task try: task = signing.SignedPackage.loads(task[1]) except (TypeError, signing.BadSignature) as e: logger.error(e) + broker.fail(ack_id) continue + task['ack_id'] = ack_id task_queue.put(task) - logger.debug(_('queueing from {}').format(list_key)) + logger.debug(_('queueing from {}').format(broker.list_key)) if event.is_set(): break logger.info(_("{} stopped pushing tasks").format(current_process().name)) -def monitor(result_queue): +def monitor(result_queue, broker=None): """ Gets finished tasks from the result queue and saves them to Django :type result_queue: multiprocessing.Queue """ + if not broker: + broker = get_broker() name = current_process().name logger.info(_("{} monitoring at {}").format(name, current_process().pid)) db.close_old_connections() for task in iter(result_queue.get, 'STOP'): + ack_id = task.pop('ack_id', False) + if ack_id: + broker.acknowledge(ack_id) save_task(task) if task['success']: logger.info(_("Processed [{}]").format(task['name'])) @@ -410,10 +418,12 @@ def save_task(task): logger.error(e) -def scheduler(list_key=Conf.Q_LIST): +def scheduler(broker=None): """ Creates a task from a schedule at the scheduled time and schedules next run """ + if not broker: + broker = get_broker() try: for s in Schedule.objects.exclude(repeats=0).filter(next_run__lt=timezone.now()): args = () @@ -456,7 +466,7 @@ def scheduler(list_key=Conf.Q_LIST): s.next_run = next_run.datetime s.repeats += -1 # send it to the cluster - q_options['list_key'] = list_key + q_options['broker'] = broker q_options['group'] = s.name or s.id kwargs['q_options'] = q_options s.task = tasks.async(s.func, *args, **kwargs) diff --git a/django_q/conf.py b/django_q/conf.py index 16a00b1..ebd2842 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -8,7 +8,6 @@ from django.conf import settings # external import os -import redis # optional try: @@ -33,6 +32,11 @@ class Conf(object): DJANGO_REDIS = conf.get('django_redis', None) + # Disque broker + DISQUE_NODES = conf.get('disque_nodes', None) + # Optional Authentication + DISQUE_AUTH = conf.get('disque_auth', None) + # Name of the cluster or site. For when you run multiple sites on one redis server PREFIX = conf.get('name', 'default') @@ -43,9 +47,6 @@ class Conf(object): # Failures are always saved SAVE_LIMIT = conf.get('save_limit', 250) - # Maximum number of tasks that each cluster can work on - QUEUE_LIMIT = conf.get('queue_limit', None) - # Number of workers in the pool. Default is cpu count if implemented, otherwise 4. WORKERS = conf.get('workers', False) if not WORKERS: @@ -60,6 +61,9 @@ class Conf(object): # sensible default WORKERS = 4 + # Maximum number of tasks that each cluster can work on + QUEUE_LIMIT = conf.get('queue_limit', int(WORKERS)**2) + # Sets compression of redis packages COMPRESSED = conf.get('compress', False) @@ -69,6 +73,10 @@ class Conf(object): # Number of seconds to wait for a worker to finish. TIMEOUT = conf.get('timeout', None) + # Number of seconds to wait for acknowledgement before retrying a task + # Only works with brokers that guarantee delivery. Defaults to 60 seconds. + RETRY = conf.get('retry', 60) + # The Django Admin label for this app LABEL = conf.get('label', 'Django Q') @@ -78,6 +86,9 @@ class Conf(object): # Global sync option to for debugging SYNC = conf.get('sync', False) + # The Django cache to use + CACHE = conf.get('cache', 'default') + # If set to False the scheduler won't execute tasks in the past. # Instead it will run once and reschedule the next run in the future. Defaults to True. CATCH_UP = conf.get('catch_up', True) @@ -86,8 +97,6 @@ class Conf(object): # Django itself should raise an error if it's not configured SECRET_KEY = settings.SECRET_KEY - # The redis list key - Q_LIST = 'django_q:{}:q'.format(PREFIX) # The redis stats key Q_STAT = 'django_q:{}:cluster'.format(PREFIX) @@ -124,28 +133,6 @@ if not logger.handlers: logger.addHandler(handler) -# Django-redis support -if Conf.DJANGO_REDIS: - try: - import django_redis - except ImportError: - django_redis = None - - -def get_redis_client(): - """ - Returns a connection from redis-py or django-redis - :return: a redis client - """ - if Conf.DJANGO_REDIS and django_redis: - return django_redis.get_redis_connection(Conf.DJANGO_REDIS) - return redis.StrictRedis(**Conf.REDIS) - - -# redis client -redis_client = get_redis_client() - - # get parent pid compatibility def get_ppid(): if hasattr(os, 'getppid'): diff --git a/django_q/monitor.py b/django_q/monitor.py index 9acc8b0..3a0dcec 100644 --- a/django_q/monitor.py +++ b/django_q/monitor.py @@ -10,14 +10,17 @@ from django.utils import timezone from django.utils.translation import ugettext as _ # local -from django_q.conf import Conf, redis_client -from django_q.status import Stat, ping_redis +from django_q.conf import Conf +from django_q.status import Stat +from django_q.brokers import get_broker from django_q import models -def monitor(run_once=False, r=redis_client): +def monitor(run_once=False, broker=None): + if not broker: + broker = get_broker() term = Terminal() - ping_redis(r) + broker.ping() with term.fullscreen(), term.hidden_cursor(), term.cbreak(): val = None start_width = int(term.width / 8) @@ -25,7 +28,7 @@ def monitor(run_once=False, r=redis_client): col_width = int(term.width / 8) # In case of resize if col_width != start_width: - print(term.clear) + print(term.clear()) start_width = col_width print(term.move(0, 0) + term.black_on_green(term.center(_('Host'), width=col_width - 1))) print(term.move(0, 1 * col_width) + term.black_on_green(term.center(_('Id'), width=col_width - 1))) @@ -36,7 +39,7 @@ def monitor(run_once=False, r=redis_client): print(term.move(0, 6 * col_width) + term.black_on_green(term.center(_('RC'), width=col_width - 1))) print(term.move(0, 7 * col_width) + term.black_on_green(term.center(_('Up'), width=col_width - 1))) i = 2 - stats = Stat.get_all(r=r) + stats = Stat.get_all(broker=broker) print(term.clear_eos()) for stat in stats: status = stat.status @@ -77,17 +80,30 @@ def monitor(run_once=False, r=redis_client): print(term.move(i, 6 * col_width) + term.center(stat.reincarnations, width=col_width - 1)) print(term.move(i, 7 * col_width) + term.center(uptime, width=col_width - 1)) i += 1 + # bottom bar + i += 1 + print(term.move(i, 0) + term.white_on_cyan(term.center(broker.info(), width=col_width * 2))) + print(term.move(i, 2 * col_width) + term.black_on_cyan(term.center(_('Queued'), width=col_width))) + print(term.move(i, 3 * col_width) + term.white_on_cyan(term.center(broker.queue_size(), width=col_width))) + print(term.move(i, 4 * col_width) + term.black_on_cyan(term.center(_('Success'), width=col_width))) + print(term.move(i, 5 * col_width) + term.white_on_cyan( + term.center(models.Success.objects.count(), width=col_width))) + print(term.move(i, 6 * col_width) + term.black_on_cyan(term.center(_('Failures'), width=col_width))) + print(term.move(i, 7 * col_width) + term.white_on_cyan( + term.center(models.Failure.objects.count(), width=col_width))) # for testing if run_once: - return Stat.get_all(r=r) + return Stat.get_all(broker=broker) print(term.move(i + 2, 0) + term.center(_('[Press q to quit]'))) val = term.inkey(timeout=1) -def info(r=redis_client): +def info(broker=None): + if not broker: + broker = get_broker() term = Terminal() - ping_redis(r) - stat = Stat.get_all(r) + broker.ping() + stat = Stat.get_all(broker=broker) # general stats clusters = len(stat) workers = 0 @@ -124,7 +140,7 @@ def info(r=redis_client): else: tasks_per = tasks_per_day # print to terminal - term.clear_eos() + print(term.clear_eos()) col_width = int(term.width / 6) print(term.black_on_green(term.center(_('-- {} summary --').format(Conf.PREFIX)))) print(term.cyan(_('Clusters')) + @@ -141,7 +157,7 @@ def info(r=redis_client): ) print(term.cyan(_('Queued')) + term.move_x(1 * col_width) + - term.white(str(r.llen(Conf.Q_LIST))) + + term.white(str(broker.queue_size())) + term.move_x(2 * col_width) + term.cyan(_('Successes')) + term.move_x(3 * col_width) + @@ -164,3 +180,4 @@ def info(r=redis_client): term.white('{0:.4f}'.format(exec_time)) ) return True + diff --git a/django_q/status.py b/django_q/status.py index ef98148..db03825 100644 --- a/django_q/status.py +++ b/django_q/status.py @@ -1,6 +1,7 @@ import socket from django.utils import timezone -from django_q.conf import Conf, logger, redis_client +from django_q.brokers import get_broker +from django_q.conf import Conf, logger import signing @@ -27,7 +28,7 @@ class Stat(Status): def __init__(self, sentinel): super(Stat, self).__init__(sentinel.parent_pid or sentinel.pid) - self.r = sentinel.r + self.broker = sentinel.broker or get_broker() self.tob = sentinel.tob self.reincarnations = sentinel.reincarnations self.sentinel = sentinel.pid @@ -63,7 +64,7 @@ class Stat(Status): def save(self): try: - self.r.set(self.key, signing.SignedPackage.dumps(self, True), 3) + self.broker.set_stat(self.key, signing.SignedPackage.dumps(self, True), 3) except Exception as e: logger.error(e) @@ -71,15 +72,16 @@ class Stat(Status): return self.done_q_size + self.task_q_size == 0 @staticmethod - def get(cluster_id, r=redis_client): + def get(cluster_id, broker=None): """ gets the current status for the cluster :param cluster_id: id of the cluster :return: Stat or Status """ - key = Stat.get_key(cluster_id) - if r.exists(key): - pack = r.get(key) + if not broker: + broker = get_broker() + pack = broker.get_stat(Stat.get_key(cluster_id)) + if pack: try: return signing.SignedPackage.loads(pack) except signing.BadSignature: @@ -87,33 +89,25 @@ class Stat(Status): return Status(cluster_id) @staticmethod - def get_all(r=redis_client): + def get_all(broker=None): """ Get the status for all currently running clusters with the same prefix and secret key. :return: list of type Stat """ + if not broker: + broker = get_broker() stats = [] - keys = r.keys(pattern='{}:*'.format(Conf.Q_STAT)) - if keys: - packs = r.mget(keys) - for pack in packs: - try: - stats.append(signing.SignedPackage.loads(pack)) - except signing.BadSignature: - continue + packs = broker.get_stats('{}:*'.format(Conf.Q_STAT)) or [] + for pack in packs: + try: + stats.append(signing.SignedPackage.loads(pack)) + except signing.BadSignature: + continue return stats def __getstate__(self): # Don't pickle the redis connection state = dict(self.__dict__) - del state['r'] + del state['broker'] return state - - -def ping_redis(r): - try: - r.ping() - except Exception as e: - logger.error('Can not connect to Redis server.') - raise e diff --git a/django_q/tasks.py b/django_q/tasks.py index b9c7360..17a8952 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -7,9 +7,10 @@ from django.utils import timezone # local import signing import cluster -from django_q.conf import Conf, redis_client, logger +from django_q.conf import Conf, logger from django_q.models import Schedule, Task from django_q.humanhash import uuid +from django_q.brokers import get_broker def async(func, *args, **kwargs): @@ -17,8 +18,7 @@ def async(func, *args, **kwargs): # get options from q_options dict or direct from kwargs options = kwargs.pop('q_options', kwargs) hook = options.pop('hook', None) - list_key = options.pop('list_key', Conf.Q_LIST) - redis = options.pop('redis', redis_client) + broker = options.pop('broker', get_broker()) sync = options.pop('sync', False) group = options.pop('group', None) save = options.pop('save', None) @@ -42,7 +42,7 @@ def async(func, *args, **kwargs): if sync or Conf.SYNC: return _sync(pack) # push it - redis.rpush(list_key, pack) + broker.enqueue(pack) logger.debug('Pushed {}'.format(tag)) return task['id'] @@ -152,17 +152,18 @@ def delete_group(group_id, tasks=False): return Task.delete_group(group_id, tasks) -def queue_size(list_key=Conf.Q_LIST, r=redis_client): +def queue_size(broker=None): """ Returns the current queue size. Note that this doesn't count any tasks currently being processed by workers. - :param list_key: optional redis key - :param r: optional redis connection + :param broker: optional broker :return: current queue size :rtype: int """ - return r.llen(list_key) + if not broker: + broker = get_broker() + return broker.queue_size() def _sync(pack): diff --git a/django_q/tests/tasks.py b/django_q/tests/tasks.py index 9c8f8ad..62e5674 100644 --- a/django_q/tests/tasks.py +++ b/django_q/tests/tasks.py @@ -1,4 +1,3 @@ -# simple countdown, returns nothing from time import sleep diff --git a/django_q/tests/test_brokers.py b/django_q/tests/test_brokers.py new file mode 100644 index 0000000..836e59f --- /dev/null +++ b/django_q/tests/test_brokers.py @@ -0,0 +1,85 @@ +from time import sleep +import pytest +import os +import redis +from django_q.conf import Conf +from django_q.brokers import get_broker, Broker + + +def test_broker(): + broker = Broker() + broker.enqueue('test') + broker.dequeue() + broker.queue_size() + broker.purge_queue() + broker.delete('id') + broker.delete_queue() + broker.acknowledge('test') + broker.ping() + broker.info() + assert broker.get_stat('test_1') is None + broker.set_stat('test_1', 'test', 3) + assert broker.get_stat('test_1') == 'test' + assert broker.get_stats('test:*')[0] == 'test' + + +def test_redis(): + Conf.DJANGO_REDIS = None + broker = get_broker() + assert broker.ping() is True + assert broker.info() is not None + Conf.REDIS = {'host': '127.0.0.1', 'port': 7712} + broker = get_broker() + with pytest.raises(Exception): + broker.ping() + Conf.REDIS = None + Conf.DJANGO_REDIS = 'default' + + +def test_disque(): + Conf.DISQUE_NODES = ['127.0.0.1:7711'] + # check broker + broker = get_broker(list_key='disque_test') + assert broker.ping() is True + assert broker.info() is not None + # clear before we start + broker.delete_queue() + # enqueue + broker.enqueue('test') + assert broker.queue_size() == 1 + # dequeue + task = broker.dequeue() + assert task[1] == 'test' + broker.acknowledge(task[0]) + assert broker.queue_size() == 0 + # Retry test + Conf.RETRY = 1 + broker.enqueue('test') + assert broker.queue_size() == 1 + broker.dequeue() + assert broker.queue_size() == 0 + sleep(1.5) + assert broker.queue_size() == 1 + task = broker.dequeue() + assert broker.queue_size() == 0 + broker.acknowledge(task[0]) + sleep(1.5) + assert broker.queue_size() == 0 + # connection test + Conf.DISQUE_NODES = ['127.0.0.1:7712', '127.0.0.1:7713'] + with pytest.raises(redis.exceptions.ConnectionError): + broker.get_connection() + # delete job + task_id = broker.enqueue('test') + broker.delete(task_id) + assert broker.queue_size() == 0 + # fail + task_id=broker.enqueue('test') + broker.fail(task_id) + # delete queue + broker.enqueue('test') + broker.enqueue('test') + broker.delete_queue() + assert broker.queue_size() == 0 + # back to django-redis + Conf.DISQUE_NODES = None diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 7e5a407..16d131e 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -13,8 +13,9 @@ from django_q.cluster import Cluster, Sentinel, pusher, worker, monitor from django_q.humanhash import DEFAULT_WORDLIST from django_q.tasks import fetch, fetch_group, async, result, result_group, count_group, delete_group, queue_size from django_q.models import Task, Success -from django_q.conf import Conf, redis_client +from django_q.conf import Conf from django_q.status import Stat +from django_q.brokers import get_broker from .tasks import multiply @@ -27,25 +28,25 @@ class WordClass(object): @pytest.fixture -def r(): - return redis_client +def broker(): + return get_broker() -def test_redis_connection(r): - assert r.ping() is True +def test_redis_connection(broker): + assert broker.ping() is True @pytest.mark.django_db -def test_sync(r): - task = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, redis=r, sync=True) +def test_sync(broker): + task = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, broker=broker, sync=True) assert result(task) == 1506 @pytest.mark.django_db -def test_cluster_initial(r): - list_key = 'initial_test:q' - r.delete(list_key) - c = Cluster(list_key=list_key) +def test_cluster_initial(broker): + broker.list_key = 'initial_test:q' + broker.delete_queue() + c = Cluster(broker=broker) assert c.sentinel is None assert c.stat.status == Conf.STOPPED assert c.start() > 0 @@ -59,7 +60,7 @@ def test_cluster_initial(r): assert c.stop() is True assert c.sentinel.is_alive() is False assert c.has_stopped - r.delete(list_key) + broker.delete_queue() @pytest.mark.django_db @@ -67,17 +68,17 @@ def test_sentinel(): start_event = Event() stop_event = Event() stop_event.set() - s = Sentinel(stop_event, start_event, list_key='sentinel_test:q') + s = Sentinel(stop_event, start_event, broker=get_broker('sentinel_test:q')) assert start_event.is_set() assert s.status() == Conf.STOPPED @pytest.mark.django_db -def test_cluster(r): - list_key = 'cluster_test:q' - r.delete(list_key) - task = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, list_key=list_key) - assert queue_size(list_key=list_key, r=r) == 1 +def test_cluster(broker): + broker.list_key = 'cluster_test:q' + broker.delete_queue() + task = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, broker=broker) + assert broker.queue_size() == 1 task_queue = Queue() assert task_queue.qsize() == 0 result_queue = Queue() @@ -85,9 +86,9 @@ def test_cluster(r): event = Event() event.set() # Test push - pusher(task_queue, event, list_key=list_key) + pusher(task_queue, event, broker=broker) assert task_queue.qsize() == 1 - assert queue_size(list_key=list_key, r=r) == 0 + assert queue_size(broker=broker) == 0 # Test work task_queue.put('STOP') worker(task_queue, result_queue, Value('f', -1)) @@ -99,36 +100,36 @@ def test_cluster(r): assert result_queue.qsize() == 0 # check result assert result(task) == 1506 - r.delete(list_key) + broker.delete_queue() @pytest.mark.django_db -def test_async(r, admin_user): - list_key = 'cluster_test:q' - r.delete(list_key) +def test_async(broker, admin_user): + broker.list_key = 'cluster_test:q' + broker.delete_queue() a = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, hook='django_q.tests.test_cluster.assert_result', - list_key=list_key, redis=r) + broker=broker) b = async('django_q.tests.tasks.count_letters2', WordClass(), hook='django_q.tests.test_cluster.assert_result', - list_key=list_key, redis=r) + broker=broker) # unknown argument c = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, 'oneargumentoomany', - hook='django_q.tests.test_cluster.assert_bad_result', list_key=list_key, redis=r) + hook='django_q.tests.test_cluster.assert_bad_result', broker=broker) # unknown function d = async('django_q.tests.tasks.does_not_exist', WordClass(), hook='django_q.tests.test_cluster.assert_bad_result', - list_key=list_key, redis=r) + broker=broker) # function without result - e = async('django_q.tests.tasks.countdown', 100000, list_key=list_key, redis=r) + e = async('django_q.tests.tasks.countdown', 100000, broker=broker) # function as instance - f = async(multiply, 753, 2, hook=assert_result, list_key=list_key, redis=r) + f = async(multiply, 753, 2, hook=assert_result, broker=broker) # model as argument - g = async('django_q.tests.tasks.get_task_name', Task(name='John'), list_key=list_key, redis=r) + g = async('django_q.tests.tasks.get_task_name', Task(name='John'), broker=broker) # args,kwargs, group and broken hook - h = async('django_q.tests.tasks.word_multiply', 2, word='django', hook='fail.me', list_key=list_key, redis=r) + h = async('django_q.tests.tasks.word_multiply', 2, word='django', hook='fail.me', broker=broker) # args unpickle test - j = async('django_q.tests.tasks.get_user_id', admin_user, list_key=list_key, group='test_j', redis=r) + j = async('django_q.tests.tasks.get_user_id', admin_user, broker=broker, group='test_j') # q_options and save opt_out test k = async('django_q.tests.tasks.get_user_id', admin_user, - q_options={'list_key': list_key, 'group': 'test_k', 'redis': r, 'save': False, 'timeout': 90}) + q_options={'broker': broker, 'group': 'test_k', 'save': False, 'timeout': 90}) # check if everything has a task id assert isinstance(a, str) assert isinstance(b, str) @@ -142,14 +143,14 @@ def test_async(r, admin_user): assert isinstance(k, str) # run the cluster to execute the tasks task_count = 10 - assert queue_size(list_key=list_key, r=r) == task_count + assert broker.queue_size() == task_count task_queue = Queue() stop_event = Event() stop_event.set() # push the tasks for i in range(task_count): - pusher(task_queue, stop_event, list_key=list_key) - assert queue_size(list_key=list_key, r=r) == 0 + pusher(task_queue, stop_event, broker=broker) + assert broker.queue_size() == 0 assert task_queue.qsize() == task_count task_queue.put('STOP') # let a worker handle them @@ -218,64 +219,66 @@ def test_async(r, admin_user): assert delete_group('test_j', tasks=True) is None # task k should not have been saved assert fetch(k) is None - r.delete(list_key) + broker.delete_queue() @pytest.mark.django_db -def test_timeout(r): +def test_timeout(broker): # set up the Sentinel - list_key = 'timeout_test:q' - async('django_q.tests.tasks.count_forever', list_key=list_key) + broker.list_key = 'timeout_test:q' + broker.purge_queue() + async('django_q.tests.tasks.count_forever', broker=broker) start_event = Event() stop_event = Event() # Set a timer to stop the Sentinel threading.Timer(3, stop_event.set).start() - s = Sentinel(stop_event, start_event, list_key=list_key, timeout=1) + s = Sentinel(stop_event, start_event, broker=broker, timeout=1) assert start_event.is_set() assert s.status() == Conf.STOPPED assert s.reincarnations == 1 - r.delete(list_key) + broker.delete_queue() @pytest.mark.django_db -def test_timeout(r): +def test_timeout(broker): # set up the Sentinel - list_key = 'timeout_test:q' - async('django_q.tests.tasks.count_forever', list_key=list_key) + broker.list_key = 'timeout_test:q' + broker.purge_queue() + async('django_q.tests.tasks.count_forever', broker=broker) start_event = Event() stop_event = Event() # Set a timer to stop the Sentinel threading.Timer(3, stop_event.set).start() - s = Sentinel(stop_event, start_event, list_key=list_key, timeout=1) + s = Sentinel(stop_event, start_event, broker=broker, timeout=1) assert start_event.is_set() assert s.status() == Conf.STOPPED assert s.reincarnations == 1 - r.delete(list_key) + broker.delete_queue() @pytest.mark.django_db -def test_timeout_override(r): +def test_timeout_override(broker): # set up the Sentinel - list_key = 'timeout_override_test:q' - async('django_q.tests.tasks.count_forever', list_key=list_key, timeout=1) + broker.list_key = 'timeout_override_test:q' + async('django_q.tests.tasks.count_forever', broker=broker, timeout=1) start_event = Event() stop_event = Event() # Set a timer to stop the Sentinel threading.Timer(3, stop_event.set).start() - s = Sentinel(stop_event, start_event, list_key=list_key, timeout=10) + s = Sentinel(stop_event, start_event, broker=broker, timeout=10) assert start_event.is_set() assert s.status() == Conf.STOPPED assert s.reincarnations == 1 - r.delete(list_key) + broker.delete_queue() @pytest.mark.django_db -def test_recycle(r): +def test_recycle(broker): # set up the Sentinel - list_key = 'test_recycle_test:q' - async('django_q.tests.tasks.multiply', 2, 2, list_key=list_key, redis=r) - async('django_q.tests.tasks.multiply', 2, 2, list_key=list_key, redis=r) - async('django_q.tests.tasks.multiply', 2, 2, list_key=list_key, redis=r) + broker.list_key = 'test_recycle_test:q' + async('django_q.tests.tasks.multiply', 2, 2, broker=broker) + async('django_q.tests.tasks.multiply', 2, 2, broker=broker) + async('django_q.tests.tasks.multiply', 2, 2, broker=broker) start_event = Event() stop_event = Event() # override settings @@ -283,17 +286,17 @@ def test_recycle(r): Conf.WORKERS = 1 # set a timer to stop the Sentinel threading.Timer(3, stop_event.set).start() - s = Sentinel(stop_event, start_event, list_key=list_key) + s = Sentinel(stop_event, start_event, broker=broker) assert start_event.is_set() assert s.status() == Conf.STOPPED assert s.reincarnations == 1 - async('django_q.tests.tasks.multiply', 2, 2, list_key=list_key, redis=r) - async('django_q.tests.tasks.multiply', 2, 2, list_key=list_key, redis=r) + async('django_q.tests.tasks.multiply', 2, 2, broker=broker) + async('django_q.tests.tasks.multiply', 2, 2, broker=broker) task_queue = Queue() result_queue = Queue() # push two tasks - pusher(task_queue, stop_event, list_key=list_key) - pusher(task_queue, stop_event, list_key=list_key) + pusher(task_queue, stop_event, broker=broker) + pusher(task_queue, stop_event, broker=broker) # worker should exit on recycle worker(task_queue, result_queue, Value('f', -1)) # check if the work has been done @@ -304,30 +307,30 @@ def test_recycle(r): # run monitor monitor(result_queue) assert Success.objects.count() == Conf.SAVE_LIMIT - r.delete(list_key) + broker.delete_queue() @pytest.mark.django_db -def test_bad_secret(r, monkeypatch): - list_key = 'test_bad_secret' - async('math.copysign', 1, -1, list_key=list_key) +def test_bad_secret(broker, monkeypatch): + broker.list_key = 'test_bad_secret:q' + async('math.copysign', 1, -1, broker=broker) stop_event = Event() stop_event.set() start_event = Event() - s = Sentinel(stop_event, start_event, list_key=list_key, start=False) + s = Sentinel(stop_event, start_event, broker=broker, start=False) Stat(s).save() # change the SECRET monkeypatch.setattr(Conf, "SECRET_KEY", "OOPS") - stat = Stat.get_all(r) + stat = Stat.get_all() assert len(stat) == 0 - assert Stat.get(s.parent_pid, r) is None + assert Stat.get(s.parent_pid) is None task_queue = Queue() - pusher(task_queue, stop_event, list_key=list_key) + pusher(task_queue, stop_event, broker=broker) result_queue = Queue() task_queue.put('STOP') worker(task_queue, result_queue, Value('f', -1), ) assert result_queue.qsize() == 0 - r.delete(list_key) + broker.delete_queue() @pytest.mark.django_db diff --git a/django_q/tests/test_config.py b/django_q/tests/test_config.py deleted file mode 100644 index de5ab71..0000000 --- a/django_q/tests/test_config.py +++ /dev/null @@ -1,15 +0,0 @@ -import pytest - -from django_q import conf - - -@pytest.fixture -def r(): - return conf.redis_client - - -def test_django_redis(): - conf.Conf.DJANGO_REDIS = None - assert conf.redis_client.ping() is True - conf.Conf.DJANGO_REDIS = 'default' - assert conf.redis_client.ping() is True diff --git a/django_q/tests/test_monitor.py b/django_q/tests/test_monitor.py index 4f94b75..e1a3ac0 100644 --- a/django_q/tests/test_monitor.py +++ b/django_q/tests/test_monitor.py @@ -1,10 +1,9 @@ import pytest -import redis from django_q import async from django_q.cluster import Cluster from django_q.monitor import monitor, info -from django_q.status import Stat, ping_redis +from django_q.status import Stat @pytest.mark.django_db @@ -37,10 +36,3 @@ def test_info(): def do_sync(): async('django_q.tests.tasks.countdown', 1, sync=True, save=True) - - -@pytest.mark.django_db -def test_ping_redis(): - r = redis.StrictRedis(port=6388) - with pytest.raises(Exception): - ping_redis(r) diff --git a/django_q/tests/test_scheduler.py b/django_q/tests/test_scheduler.py index 7709477..2ee337a 100644 --- a/django_q/tests/test_scheduler.py +++ b/django_q/tests/test_scheduler.py @@ -6,20 +6,21 @@ import arrow from django.utils import timezone -from django_q.conf import redis_client, Conf +from django_q.brokers import get_broker +from django_q.conf import Conf from django_q.cluster import pusher, worker, monitor, scheduler from django_q.tasks import Schedule, fetch, schedule as create_schedule, queue_size @pytest.fixture -def r(): - return redis_client +def broker(): + return get_broker() @pytest.mark.django_db -def test_scheduler(r): - list_key = 'scheduler_test:q' - r.delete(list_key) +def test_scheduler(broker): + broker.list_key = 'scheduler_test:q' + broker.delete_queue() schedule = create_schedule('math.copysign', 1, -1, name='test math', @@ -28,15 +29,15 @@ def test_scheduler(r): repeats=1) assert schedule.last_run() is None # run scheduler - scheduler(list_key=list_key) + scheduler(broker=broker) # set up the workflow task_queue = Queue() stop_event = Event() stop_event.set() # push it - pusher(task_queue, stop_event, list_key=list_key) + pusher(task_queue, stop_event, broker=broker) assert task_queue.qsize() == 1 - assert queue_size(list_key=list_key, r=r) == 0 + assert broker.queue_size() == 0 task_queue.put('STOP') # let a worker handle them result_queue = Queue() @@ -91,7 +92,7 @@ def test_scheduler(r): ) assert schedule is not None assert schedule.last_run() is None - scheduler(list_key=list_key) + scheduler(broker=broker) # via model Schedule.objects.create(func='django_q.tests.tasks.word_multiply', args='2', @@ -99,7 +100,7 @@ def test_scheduler(r): schedule_type=Schedule.DAILY ) # scheduler - scheduler(list_key=list_key) + scheduler(broker=broker) # ONCE schedule should be deleted assert Schedule.objects.filter(pk=once_schedule.pk).exists() is False # Catch up On @@ -112,13 +113,13 @@ def test_scheduler(r): next_run=timezone.now() - timedelta(hours=12), repeats=-1 ) - scheduler(list_key=list_key) + scheduler(broker=broker) schedule = Schedule.objects.get(pk=schedule.pk) assert schedule.next_run < now # Catch up off Conf.CATCH_UP = False - scheduler(list_key=list_key) + scheduler(broker=broker) schedule = Schedule.objects.get(pk=schedule.pk) assert schedule.next_run > now # Done - r.delete(list_key) + broker.delete_queue() diff --git a/docs/_static/cluster.png b/docs/_static/cluster.png index 69fb38f..d7b9b87 100644 Binary files a/docs/_static/cluster.png and b/docs/_static/cluster.png differ diff --git a/docs/brokers.rst b/docs/brokers.rst new file mode 100644 index 0000000..c285e15 --- /dev/null +++ b/docs/brokers.rst @@ -0,0 +1,82 @@ +Brokers +======= + +The broker sits between your Django instances and your Django Q cluster instances, accepting and delivering task packages. +Currently we only support `Redis `__ and `Disque `__, but support for other brokers is being worked on. + +Clients for `Amazon SQS `__ and `IronMQ `__ are TBA. + + +Redis +----- +The default broker for Django Q clusters. + +* Atomic +* Does not need separate cache framework for monitoring +* Does not support receipts +* Requires `Redis-py `__ client library: ``pip install redis`` +* Can use existing :ref:`django_redis` connections. +* Configure with :ref:`redis_configuration`-py compatible configuration + +Disque +------ +Unlike Redis, Disque supports message receipts which make delivery to the cluster workers guaranteed. If a task never produces a failed or successful result, it will automatically be sent to the cluster again for a retry. +You can control the amount of time Disque should wait for completion of a task by configuring the :ref:`retry` setting. + +* Delivery receipts +* Atomic +* Needs Django's `Cache framework `__ configured for monitoring +* Compatible with `Tynd `__ Disque addon on `Heroku `__ +* Still considered Alpha software +* Requires `Redis-py `__ client library: ``pip install redis`` +* See the :ref:`disque_configuration` configuration section for more info. + +Reference +--------- +The :class:`Broker` class is used internally to communicate with the different types of brokers. +You can override this class if you want to contribute and support your own broker. + +.. py:class:: Broker + + .. py:method:: enqueue(task) + + Sends a task package to the broker queue and returns a tracking id. + + .. py:method:: dequeue() + + Gets a task package from the broker. + + .. py:method:: acknowledge(id) + + Notifies the broker that the task has been processed. + Only works with brokers that support delivery receipts. + + .. py:method:: fail(id) + + Tells the broker that the message failed to be processed by the cluster. + Only available on brokers that support this. + Currently only occurs when a cluster fails to unpack a task package. + + .. py:method:: delete(id) + + Instructs the broker to delete this message from the queue. + + .. py:method:: purge_queue() + + Empties the current queue of all messages. + + .. py:method:: delete_queue() + + Deletes the current queue from the broker. + + .. py:method:: queue_size() + + Returns the amount of messages in the brokers queue. + + .. py:method:: ping() + + Returns True if the broker can be reached. + + .. py:method:: info() + + Shows the name and version of the currently configured broker. diff --git a/docs/cluster.rst b/docs/cluster.rst index 4911889..278521a 100644 --- a/docs/cluster.rst +++ b/docs/cluster.rst @@ -48,7 +48,7 @@ Multiple Clusters ----------------- You can have multiple clusters on multiple machines, working on the same queue as long as: -- They connect to the same Redis server or Redis cluster. +- They connect to the same :doc:`broker`. - They use the same cluster name. See :doc:`configure` - They share the same ``SECRET_KEY`` for Django. @@ -92,15 +92,16 @@ Architecture Signed Tasks """""""""""" -Tasks are first pickled and then signed using Django's own :mod:`django.core.signing` module using the ``SECRET_KEY`` and cluster name as salt, before being sent to a Redis list. This ensures that task -packages on the Redis server can only be executed and read by clusters +Tasks are first pickled and then signed using Django's own :mod:`django.core.signing` module using the ``SECRET_KEY`` and cluster name as salt, before being sent to a message broker. This ensures that task +packages on the broker can only be executed and read by clusters and django servers who share the same secret key and cluster name. -Optionally the packages can be compressed before transport +If a package fails to unpack, it will be marked failed with the broker and discarded. +Optionally the packages can be compressed before transport. Pusher """""" -The pusher process continuously checks the Redis list for new task +The pusher process continuously checks the broker for new task packages. It checks the signing and unpacks the task to the Task Queue. Worker @@ -115,6 +116,7 @@ Monitor The result monitor checks the Result Queue for processed packages and saves both failed and successful packages to the Django database. +If the broker supports it, a delivery receipt is sent. .. _sentinel: diff --git a/docs/conf.py b/docs/conf.py index 4775ee9..4140e5c 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -70,9 +70,9 @@ author = 'Ilan Steemers' # built documents. # # The short X.Y version. -version = '0.5' +version = '0.6' # The full version, including alpha/beta/rc tags. -release = '0.5.3' +release = '0.6.0' # The language for content autogenerated by Sphinx. Refer to documentation # for a list of supported languages. diff --git a/docs/configure.rst b/docs/configure.rst index fe76ffe..423af59 100644 --- a/docs/configure.rst +++ b/docs/configure.rst @@ -52,6 +52,14 @@ timeout The number of seconds a worker is allowed to spend on a task before it's terminated. Defaults to ``None``, meaning it will never time out. Set this to something that makes sense for your project. Can be overridden for individual tasks. +.. _retry: + +retry +~~~~~ + +The number of seconds a broker will wait for a cluster to finish a task, before it's presented again. +Only works with brokers that support delivery receipts. Defaults to 60 seconds. + compress ~~~~~~~~ @@ -82,10 +90,9 @@ Effectively making everything synchronous. Useful for testing. Defaults to ``Fal queue_limit ~~~~~~~~~~~ -This does not limit the amount of tasks that can be queued overall on Redis, but rather how many tasks are kept in memory by a single cluster. +This does not limit the amount of tasks that can be queued on the broker, but rather how many tasks are kept in memory by a single cluster. Setting this to a reasonable number, can help balance the workload and the memory overhead of each individual cluster. -It can also be used to manage the loss of data in case of a cluster failure. -Defaults to ``None``, meaning no limit. +Defaults to ``workers**2``. label ~~~~~ @@ -100,20 +107,25 @@ The default behavior for schedules that didn't run while a cluster was down, is You can override this behavior by setting ``catch_up`` to ``False``. This will make those schedules run only once when the cluster starts and normal scheduling resumes. Defaults to ``True``. +.. _redis_configuration: + redis ~~~~~ Connection settings for Redis. Defaults:: - redis: { - 'host': 'localhost', - 'port': 6379, - 'db': 0, - 'password': None, - 'socket_timeout': None, - 'charset': 'utf-8', - 'errors': 'strict', - 'unix_socket_path': None + # redis defaults + Q_CLUSTER = { + 'redis': { + 'host': 'localhost', + 'port': 6379, + 'db': 0, + 'password': None, + 'socket_timeout': None, + 'charset': 'utf-8', + 'errors': 'strict', + 'unix_socket_path': None + } } For more information on these settings please refer to the `Redis-py `__ documentation @@ -131,7 +143,7 @@ of the cache connection you want to use:: 'name': 'DJRedis', 'workers': 4, 'timeout': 90, - 'django_redis: 'default' + 'django_redis': 'default' } @@ -139,6 +151,42 @@ of the cache connection you want to use:: .. tip:: Django Q uses your ``SECRET_KEY`` to encrypt task packages and prevent task crossover. So make sure you have it set up in your Django settings. +.. _disque_configuration: + +disque_nodes +~~~~~~~~~~~~ +If you want to use Disque as your broker, set this to a list of available Disque nodes and each cluster will randomly try to connect to them:: + + # example disque connection + Q_CLUSTER = { + 'name': 'DisqueBroker', + 'workers': 4, + 'timeout': 60, + 'retry': 60, + 'disque_nodes': ['127.0.0.1:7711', '127.0.0.1:7712'] + } + + +Django Q is also compatible with the `Tynd `__ addon on `Heroku `__:: + + # example Tynd connection + import os + + Q_CLUSTER = { + 'name': 'TyndBroker', + 'workers': 8, + 'timeout': 30, + 'retry': 60, + 'disque_nodes': os.environ['TYND_DISQUE_NODES'].split(','), + 'disque_auth': os.environ['TYND_DISQUE_AUTH'] + } + + +disque_auth +~~~~~~~~~~~ + +Optional Disque password for servers that require authentication. + cpu_affinity ~~~~~~~~~~~~ diff --git a/docs/index.rst b/docs/index.rst index 3bad6a1..5d38a22 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -20,7 +20,7 @@ Features - Django Admin integration - PaaS compatible with multiple instances - Multi cluster monitor -- Redis broker +- Redis or Disque broker - Python 2 and 3 @@ -33,6 +33,7 @@ Contents: Installation Configuration + Brokers Tasks Schedules Cluster diff --git a/docs/install.rst b/docs/install.rst index 1297921..fa750be 100644 --- a/docs/install.rst +++ b/docs/install.rst @@ -18,8 +18,8 @@ Installation $ python manage.py migrate -- Make sure you have a `Redis `__ server running - somewhere and know how to connect to it. +- Choose a message :doc:`broker` , configure it and install the appropriate client library. + Requirements ------------ @@ -35,10 +35,6 @@ Django Q is tested for Python 2.7 and 3.4 Used to store args, kwargs and result objects in the database. -- `Redis-py `__ - - Andy McCurdy's excellent Redis python client. - - `Arrow `__ The scheduler uses Chris Smith's wonderful project to determine correct dates in the future. @@ -47,21 +43,25 @@ Django Q is tested for Python 2.7 and 3.4 This feature-filled fork of Erik Rose's blessings project provides the terminal layout of the monitor. -- `Redis server `__ - - Django Q uses Redis as a centralized hub between your Django instances and your Q clusters. - Optional ~~~~~~~~ +- `Redis-py `__ client by Andy McCurdy is used to interface with both the Redis and Disque brokers:: + + $ pip install redis + .. _psutil: - `Psutil `__ python system and process utilities module by Giampaolo Rodola', is an optional requirement and adds cpu affinity settings to the cluster:: $ pip install psutil - - `Hiredis `__ parser. This C library maintained by the core Redis team is faster than the standard PythonParser during high loads:: $ pip install hiredis +- `Redis `__ server is the default broker for Django Q. It provides the best performance and does not require Django's cache framework for monitoring. + +- `Disque `__ server is based on Redis by the same author, but focuses on reliable queues. Currently in Alpha, but highly recommended. You can either build it from source or use it on Heroku through the `Tynd `__ beta. + + diff --git a/requirements.txt b/requirements.txt index 83e7e99..1c376c0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -11,7 +11,7 @@ django-redis==4.2.0 future==0.15.0 hiredis==0.2.0 msgpack-python==0.4.6 # via django-redis -psutil==3.1.1 +psutil==3.2.1 python-dateutil==2.4.2 # via arrow redis==2.10.3 six==1.9.0 # via django-picklefield, python-dateutil diff --git a/setup.py b/setup.py index aad8571..1631e60 100644 --- a/setup.py +++ b/setup.py @@ -26,17 +26,17 @@ class PyTest(Command): setup( name='django-q', - version='0.5.3', + version='0.6.0', author='Ilan Steemers', author_email='koed00@gmail.com', - keywords='django task queue worker redis multiprocessing', + keywords='django task queue worker redis disque multiprocessing', packages=['django_q'], include_package_data=True, url='https://django-q.readthedocs.org', license='MIT', description='A multiprocessing task queue for Django', long_description=README, - install_requires=['django>=1.7', 'redis', 'django-picklefield', 'blessed', 'arrow', 'future'], + install_requires=['django>=1.7', 'django-picklefield', 'blessed', 'arrow', 'future'], test_requires=['pytest', 'pytest-django', ], cmdclass={'test': PyTest}, classifiers=[