From 54d2822f5c035d9542c769e2c3dc2e93553b9ef7 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Wed, 9 Sep 2015 15:21:13 +0200 Subject: [PATCH 01/10] updates future to 0.15.1 --- requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index 4859074..7b03a84 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,7 +8,7 @@ arrow==0.6.0 blessed==1.9.5 django-picklefield==0.3.2 django-redis==4.2.0 -future==0.15.0 +future==0.15.1 hiredis==0.2.0 iron-core==1.1.9 # via iron-mq iron-mq==0.7 From d7ce945f0374b16137655c4c31125b52aade483a Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Wed, 9 Sep 2015 18:10:03 +0200 Subject: [PATCH 02/10] Amazon SQS broker --- django_q/brokers/__init__.py | 3 ++ django_q/brokers/aws_sqs.py | 69 ++++++++++++++++++++++++++++++++++ django_q/conf.py | 3 ++ django_q/tests/test_brokers.py | 55 +++++++++++++++++++++++++++ requirements.in | 1 + requirements.txt | 1 + 6 files changed, 132 insertions(+) create mode 100644 django_q/brokers/aws_sqs.py diff --git a/django_q/brokers/__init__.py b/django_q/brokers/__init__.py index 626717c..fed4590 100644 --- a/django_q/brokers/__init__.py +++ b/django_q/brokers/__init__.py @@ -157,6 +157,9 @@ def get_broker(list_key=Conf.PREFIX): elif Conf.IRON_MQ: from brokers import ironmq return ironmq.IronMQBroker(list_key=list_key) + elif Conf.SQS: + from brokers import aws_sqs + return aws_sqs.Sqs(list_key=list_key) # default to redis else: from brokers import redis_broker diff --git a/django_q/brokers/aws_sqs.py b/django_q/brokers/aws_sqs.py new file mode 100644 index 0000000..93bc16e --- /dev/null +++ b/django_q/brokers/aws_sqs.py @@ -0,0 +1,69 @@ +from django_q.conf import Conf +from django_q.brokers import Broker +import boto.sqs +from boto.sqs.message import RawMessage + + +class Sqs(Broker): + def __init__(self, list_key=Conf.PREFIX): + super(Sqs, self).__init__(list_key) + self.queue = self.get_queue() + + def enqueue(self, task): + m = RawMessage() + m.set_body(task) + self.queue.write(m) + return m.id + + def dequeue(self): + t = None + if len(self.task_cache) > 0: + t = self.task_cache.pop() + else: + tasks = self.queue.get_messages(num_messages=Conf.BULK, visibility_timeout=Conf.RETRY) + if tasks: + t = tasks.pop() + if tasks: + self.task_cache = tasks + if t: + return t.receipt_handle, t.get_body() + + def acknowledge(self, task_id): + return self.delete(task_id) + + def queue_size(self): + return self.queue.count() + + def delete(self, task_id): + m = RawMessage() + m.receipt_handle = task_id + return self.queue.delete_message(m) + + def fail(self, task_id): + self.delete(task_id) + + def delete_queue(self): + self.connection.delete_queue(self.queue) + + def purge_queue(self): + self.queue.purge() + + def ping(self): + try: + self.connection.get_all_queues() + return True + except Exception as e: + raise e + + def info(self): + return 'AWS SQS' + + @staticmethod + def get_connection(list_key=Conf.PREFIX): + conn = boto.sqs.connect_to_region(Conf.SQS['aws_region'], + aws_access_key_id=Conf.SQS['aws_access_key_id'], + aws_secret_access_key=Conf.SQS['aws_secret_access_key']) + return conn + + def get_queue(self): + return self.connection.create_queue(self.list_key) diff --git a/django_q/conf.py b/django_q/conf.py index 9463806..c723fac 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -40,6 +40,9 @@ class Conf(object): # IronMQ broker IRON_MQ = conf.get('iron_mq', None) + # SQS broker + SQS = conf.get('sqs', None) + # Name of the cluster or site. For when you run multiple sites on one redis server PREFIX = conf.get('name', 'default') diff --git a/django_q/tests/test_brokers.py b/django_q/tests/test_brokers.py index 3086613..97c6e86 100644 --- a/django_q/tests/test_brokers.py +++ b/django_q/tests/test_brokers.py @@ -148,3 +148,58 @@ def test_ironmq(): # back to django-redis Conf.IRON_MQ = None Conf.DJANGO_REDIS = 'default' + + +@pytest.mark.skipif(not os.getenv('AWS_ACCESS_KEY_ID'), + reason="requires AWS credentials") +def test_sqs(): + Conf.SQS = {'aws_region': os.getenv('AWS_REGION'), + 'aws_access_key_id': os.getenv('AWS_ACCESS_KEY_ID'), + 'aws_secret_access_key': os.getenv('AWS_SECRET_ACCESS_KEY')} + # check broker + broker = get_broker(list_key=uuid()[0]) + assert broker.ping() is True + assert broker.info() is not None + assert broker.queue_size() == 0 + # enqueue + broker.enqueue('test') + # dequeue + task = broker.dequeue() + assert task[1] == 'test' + broker.acknowledge(task[0]) + assert broker.dequeue() is None + # Retry test + Conf.RETRY = 1 + broker.enqueue('test') + assert broker.dequeue() is not None + sleep(1.5) + task = broker.dequeue() + assert len(task) > 0 + broker.acknowledge(task[0]) + sleep(1.5) + # delete job + broker.enqueue('test') + task_id = broker.dequeue()[0] + broker.delete(task_id) + assert broker.dequeue() is None + # fail + broker.enqueue('test') + task_id = broker.dequeue()[0] + broker.fail(task_id) + # bulk test + for i in range(5): + broker.enqueue('test') + Conf.BULK = 5 + for i in range(5): + task = broker.dequeue() + assert task is not None + broker.acknowledge(task[0]) + # delete queue + broker.enqueue('test') + broker.enqueue('test') + broker.purge_queue() + assert broker.dequeue() is None + broker.delete_queue() + # back to django-redis + Conf.SQS = None + Conf.DJANGO_REDIS = 'default' diff --git a/requirements.in b/requirements.in index c302143..956f3c1 100644 --- a/requirements.in +++ b/requirements.in @@ -7,3 +7,4 @@ redis psutil django-redis iron-mq +boto diff --git a/requirements.txt b/requirements.txt index 7b03a84..ffb54c0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,6 +6,7 @@ # arrow==0.6.0 blessed==1.9.5 +boto==2.38.0 django-picklefield==0.3.2 django-redis==4.2.0 future==0.15.1 From e4a5951f12eb7b532d453112a6cc974b17426465 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Wed, 9 Sep 2015 18:59:24 +0200 Subject: [PATCH 03/10] Resets SQS conf in case of test failure --- django_q/tests/test_cluster.py | 2 ++ django_q/tests/test_scheduler.py | 4 ++++ 2 files changed, 6 insertions(+) diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 3aa09c6..dcd5394 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -18,6 +18,7 @@ from django_q.status import Stat from django_q.brokers import get_broker from .tasks import multiply + class WordClass(object): def __init__(self): self.word_list = DEFAULT_WORDLIST @@ -30,6 +31,7 @@ class WordClass(object): def broker(): Conf.DISQUE_NODES = None Conf.IRON_MQ = None + Conf.SQS = None Conf.DJANGO_REDIS = 'default' return get_broker() diff --git a/django_q/tests/test_scheduler.py b/django_q/tests/test_scheduler.py index 2ee337a..4cf7fd3 100644 --- a/django_q/tests/test_scheduler.py +++ b/django_q/tests/test_scheduler.py @@ -14,6 +14,10 @@ from django_q.tasks import Schedule, fetch, schedule as create_schedule, queue_s @pytest.fixture def broker(): + Conf.DISQUE_NODES = None + Conf.IRON_MQ = None + Conf.SQS = None + Conf.DJANGO_REDIS = 'default' return get_broker() From 0fe6c4cbdefcc25389463feeae17d21434bbfa98 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Wed, 9 Sep 2015 19:34:06 +0200 Subject: [PATCH 04/10] SQS only supports max 10 messages in bulk mode * capped bulk at 10 * added tests --- django_q/brokers/aws_sqs.py | 3 +++ django_q/tests/test_brokers.py | 7 ++++--- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/django_q/brokers/aws_sqs.py b/django_q/brokers/aws_sqs.py index 93bc16e..1caa13a 100644 --- a/django_q/brokers/aws_sqs.py +++ b/django_q/brokers/aws_sqs.py @@ -16,6 +16,9 @@ class Sqs(Broker): return m.id def dequeue(self): + # sqs supports max 10 messages in bulk + if Conf.BULK > 10: + Conf.BULK = 10 t = None if len(self.task_cache) > 0: t = self.task_cache.pop() diff --git a/django_q/tests/test_brokers.py b/django_q/tests/test_brokers.py index 97c6e86..9e0b4b0 100644 --- a/django_q/tests/test_brokers.py +++ b/django_q/tests/test_brokers.py @@ -187,10 +187,10 @@ def test_sqs(): task_id = broker.dequeue()[0] broker.fail(task_id) # bulk test - for i in range(5): + for i in range(10): broker.enqueue('test') - Conf.BULK = 5 - for i in range(5): + Conf.BULK = 12 + for i in range(10): task = broker.dequeue() assert task is not None broker.acknowledge(task[0]) @@ -202,4 +202,5 @@ def test_sqs(): broker.delete_queue() # back to django-redis Conf.SQS = None + Conf.BULK = 1 Conf.DJANGO_REDIS = 'default' From 0b9df26218b950cd768e3434d76026c97cc15e01 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Thu, 10 Sep 2015 11:08:33 +0200 Subject: [PATCH 05/10] Converted SQS broker to Boto3 --- django_q/brokers/aws_sqs.py | 39 +++++++++++++++---------------------- requirements.in | 2 +- requirements.txt | 8 ++++++-- 3 files changed, 23 insertions(+), 26 deletions(-) diff --git a/django_q/brokers/aws_sqs.py b/django_q/brokers/aws_sqs.py index 1caa13a..db25215 100644 --- a/django_q/brokers/aws_sqs.py +++ b/django_q/brokers/aws_sqs.py @@ -1,19 +1,17 @@ from django_q.conf import Conf from django_q.brokers import Broker -import boto.sqs -from boto.sqs.message import RawMessage +from boto3 import Session class Sqs(Broker): def __init__(self, list_key=Conf.PREFIX): + self.sqs = None super(Sqs, self).__init__(list_key) self.queue = self.get_queue() def enqueue(self, task): - m = RawMessage() - m.set_body(task) - self.queue.write(m) - return m.id + response = self.queue.send_message(MessageBody=task) + return response.get('MessageId') def dequeue(self): # sqs supports max 10 messages in bulk @@ -23,50 +21,45 @@ class Sqs(Broker): if len(self.task_cache) > 0: t = self.task_cache.pop() else: - tasks = self.queue.get_messages(num_messages=Conf.BULK, visibility_timeout=Conf.RETRY) + tasks = self.queue.receive_messages(MaxNumberOfMessages=Conf.BULK, VisibilityTimeout=Conf.RETRY) if tasks: t = tasks.pop() if tasks: self.task_cache = tasks if t: - return t.receipt_handle, t.get_body() + return t.receipt_handle, t.body def acknowledge(self, task_id): return self.delete(task_id) def queue_size(self): - return self.queue.count() + return int(self.queue.attributes['ApproximateNumberOfMessages']) def delete(self, task_id): - m = RawMessage() - m.receipt_handle = task_id - return self.queue.delete_message(m) + message = self.sqs.Message(self.queue.url, task_id) + message.delete() def fail(self, task_id): self.delete(task_id) def delete_queue(self): - self.connection.delete_queue(self.queue) + self.queue.delete() def purge_queue(self): self.queue.purge() def ping(self): - try: - self.connection.get_all_queues() - return True - except Exception as e: - raise e + return 'sqs' in self.connection.get_available_resources() def info(self): return 'AWS SQS' @staticmethod def get_connection(list_key=Conf.PREFIX): - conn = boto.sqs.connect_to_region(Conf.SQS['aws_region'], - aws_access_key_id=Conf.SQS['aws_access_key_id'], - aws_secret_access_key=Conf.SQS['aws_secret_access_key']) - return conn + return Session(aws_access_key_id=Conf.SQS['aws_access_key_id'], + aws_secret_access_key=Conf.SQS['aws_secret_access_key'], + region_name=Conf.SQS['aws_region']) def get_queue(self): - return self.connection.create_queue(self.list_key) + self.sqs = self.connection.resource('sqs') + return self.sqs.create_queue(QueueName=self.list_key) diff --git a/requirements.in b/requirements.in index 956f3c1..476c4a9 100644 --- a/requirements.in +++ b/requirements.in @@ -7,4 +7,4 @@ redis psutil django-redis iron-mq -boto +boto3 diff --git a/requirements.txt b/requirements.txt index ffb54c0..daa05ec 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,16 +6,20 @@ # arrow==0.6.0 blessed==1.9.5 -boto==2.38.0 +boto3==1.1.3 +botocore==1.2.0 # via boto3 django-picklefield==0.3.2 django-redis==4.2.0 +docutils==0.12 # via botocore future==0.15.1 +futures==2.2.0 # via boto3 hiredis==0.2.0 iron-core==1.1.9 # via iron-mq iron-mq==0.7 +jmespath==0.7.1 # via boto3, botocore msgpack-python==0.4.6 # via django-redis psutil==3.2.1 -python-dateutil==2.4.2 # via arrow, iron-core +python-dateutil==2.4.2 # via arrow, botocore, iron-core redis==2.10.3 requests==2.7.0 # via iron-core six==1.9.0 # via python-dateutil From 5195464c31b360af15689cd9e0ee44fa1b209f0e Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Thu, 10 Sep 2015 11:34:35 +0200 Subject: [PATCH 06/10] tests: sqs is not atomic --- django_q/tests/test_brokers.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/django_q/tests/test_brokers.py b/django_q/tests/test_brokers.py index 9e0b4b0..68df44e 100644 --- a/django_q/tests/test_brokers.py +++ b/django_q/tests/test_brokers.py @@ -184,8 +184,9 @@ def test_sqs(): assert broker.dequeue() is None # fail broker.enqueue('test') - task_id = broker.dequeue()[0] - broker.fail(task_id) + while task is None: + task = broker.dequeue() + broker.fail(task[0]) # bulk test for i in range(10): broker.enqueue('test') @@ -196,9 +197,7 @@ def test_sqs(): broker.acknowledge(task[0]) # delete queue broker.enqueue('test') - broker.enqueue('test') broker.purge_queue() - assert broker.dequeue() is None broker.delete_queue() # back to django-redis Conf.SQS = None From 4c413fc65957c928f3a1ed9ccc58f81d8651c825 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Thu, 10 Sep 2015 13:31:37 +0200 Subject: [PATCH 07/10] Prevents crash on duplicate acknowledge --- django_q/brokers/ironmq.py | 11 +++++++++-- django_q/tests/test_brokers.py | 11 +++++++++++ 2 files changed, 20 insertions(+), 2 deletions(-) diff --git a/django_q/brokers/ironmq.py b/django_q/brokers/ironmq.py index fd47667..08bb860 100644 --- a/django_q/brokers/ironmq.py +++ b/django_q/brokers/ironmq.py @@ -1,3 +1,4 @@ +from requests.exceptions import HTTPError from django_q.conf import Conf from django_q.brokers import Broker from iron_mq import IronMQ @@ -31,13 +32,19 @@ class IronMQBroker(Broker): return self.connection.size() def delete_queue(self): - return self.connection.delete_queue()['msg'] + try: + return self.connection.delete_queue()['msg'] + except HTTPError: + return False def purge_queue(self): return self.connection.clear() def delete(self, task_id): - return self.connection.delete(task_id)['msg'] + try: + return self.connection.delete(task_id)['msg'] + except HTTPError: + return False def fail(self, task_id): self.delete(task_id) diff --git a/django_q/tests/test_brokers.py b/django_q/tests/test_brokers.py index 68df44e..9990414 100644 --- a/django_q/tests/test_brokers.py +++ b/django_q/tests/test_brokers.py @@ -85,6 +85,13 @@ def test_disque(): task = broker.dequeue() assert task is not None broker.acknowledge(task[0]) + # test duplicate acknowledge + # broker.acknowledge(task[0]) + # + # this crashes Disque when followed by a JSCAN + # https://github.com/antirez/disque/issues/113 + # confirmed fix and merge is on the way + # # delete queue broker.enqueue('test') broker.enqueue('test') @@ -139,6 +146,8 @@ def test_ironmq(): task = broker.dequeue() assert task is not None broker.acknowledge(task[0]) + # duplicate acknowledge + broker.acknowledge(task[0]) # delete queue broker.enqueue('test') broker.enqueue('test') @@ -195,6 +204,8 @@ def test_sqs(): task = broker.dequeue() assert task is not None broker.acknowledge(task[0]) + # duplicate acknowledge + broker.acknowledge(task[0]) # delete queue broker.enqueue('test') broker.purge_queue() From 9af70e799d1d9cd6c8c599cc059d4f583a63ed5b Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Thu, 10 Sep 2015 14:11:30 +0200 Subject: [PATCH 08/10] docs: Amazon SQS broker --- README.rst | 3 ++- docs/brokers.rst | 35 +++++++++++++++++++++++++++++++++-- docs/configure.rst | 28 ++++++++++++++++++++++++++++ docs/index.rst | 2 +- docs/install.rst | 12 +++++++++--- 5 files changed, 73 insertions(+), 7 deletions(-) diff --git a/README.rst b/README.rst index af1cfa5..c653a9a 100644 --- a/README.rst +++ b/README.rst @@ -20,7 +20,7 @@ Features - Django Admin integration - PaaS compatible with multiple instances - Multi cluster monitor -- Redis, Disque or IronMQ broker +- Redis, Disque, IronMQ or SQS - Python 2 and 3 Requirements @@ -38,6 +38,7 @@ Brokers - `Redis `__ - `Disque `__ - `IronMQ `__ +- `Amazon SQS `__ Installation diff --git a/docs/brokers.rst b/docs/brokers.rst index a062e06..374e2e0 100644 --- a/docs/brokers.rst +++ b/docs/brokers.rst @@ -2,7 +2,25 @@ Brokers ======= The broker sits between your Django instances and your Django Q cluster instances, accepting and delivering task packages. -Currently we support `Redis `__ , `Disque `__ and `IronMQ `__. +Currently we support a variety of brokers from the default Redis, bleeding edge Disque to the convenient Amazon SQS. + +The default Redis broker does not support message receipts. +This means that in case of a catastrophic failure of the cluster server or worker timeouts, tasks that were being executed get lost. +Keep in mind this is not the same as a failing task. If a tasks code crashes, this should only lead to a failed task status. + +Even though this might be acceptable in some use cases, you might prefer brokers with message receipts support. +These guarantee delivery by waiting for the cluster to send a receipt after the task has been processed. +In case a receipt has not been received after a set time, the task package is put back in the queue. +Django Q supports this behavior by setting the :ref:`retry` timer on brokers that support message receipts. + +Some pointers: + +* Don't set the :ref:`retry` timer to a lower or equal number than the task timeout. +* Retry time includes time the task spends waiting in the clusters internal queue. +* Don't set the :ref:`queue_limit` so high that tasks time out while waiting to be processed. +* In case a task is worked on twice, you will see a duplicate key error in the cluster logs. +* Duplicate tasks do generate additional receipt messages, but the result is discarded in favor of the first result. + Support for more brokers is being worked on. @@ -41,6 +59,19 @@ This HTTP based queue service is both available directly via `Iron.io `__ client library: ``pip install iron-mq`` * See the :ref:`ironmq_configuration` configuration section for options. +Amazon SQS +---------- +Amazon's Simple Queue Service is another HTTP based message queue. +Although `SQS `__ is not the fastest, it is stable, cheap and convenient if you already use AWS. + +* Delivery receipts +* Maximum message size is 256Kb +* Supports bulk dequeue up to 10 messages with a maximum total size of 256Kb +* Needs Django's `Cache framework `__ configured for monitoring +* Requires the `boto3 `__ client library: ``pip install boto3`` +* See the :ref:`sqs_configuration` configuration section for options. + + Reference --------- The :class:`Broker` class is used internally to communicate with the different types of brokers. @@ -54,7 +85,7 @@ You can override this class if you want to contribute and support your own broke .. py:method:: dequeue() - Gets a task package from the broker. + Gets a task package from the broker and returns a tuple with a tracking id and the package. .. py:method:: acknowledge(id) diff --git a/docs/configure.rst b/docs/configure.rst index d50aa66..28daf53 100644 --- a/docs/configure.rst +++ b/docs/configure.rst @@ -213,6 +213,34 @@ Connection settings for IronMQ:: All connection keywords are supported. See the `iron-mq `__ library for more info +.. _sqs_configuration: + +sqs +~~~ +To use Amazon SQS as a broker you need to provide the AWS region and credentials:: + + # example SQS broker connection + + Q_CLUSTER = { + 'name': 'SQSExample', + 'workers': 4, + 'timeout': 60, + 'retry': 90, + 'queue_limit': 100, + 'bulk': 5, + 'sqs': { + 'aws_region': 'us-east-1', + 'aws_access_key_id': 'ac-Idr.....YwflZBaaxI', + 'aws_secret_access_key': '500f7b....b0f302e9' + } + } + + +Please make sure these credentials have proper SQS access. + +Amazon SQS only supports a bulk setting between 1 and 10, with the total payload not exceeding 256kb. + + bulk ~~~~ Sets the number of messages each cluster tries to get from the broker per call. Setting this on supported brokers can improve performance. diff --git a/docs/index.rst b/docs/index.rst index fb698c7..4fc8886 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, Disque or IronMQ broker +- Redis, Disque, IronMQ or SQS - Python 2 and 3 diff --git a/docs/install.rst b/docs/install.rst index fa750be..d65f41f 100644 --- a/docs/install.rst +++ b/docs/install.rst @@ -60,8 +60,14 @@ Optional $ 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. +- `Boto3 `__ is used for the Amazon SQS broker in favor of the now deprecating boto library:: + + $ pip install boto3 + +- `Iron-mq `_ is the official python binding for the IronMQ broker:: + + $ pip install iron-mq + +- `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. - - From f87bb5f8a809d3a306e744f02ba329d96dddc476 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Thu, 10 Sep 2015 14:26:55 +0200 Subject: [PATCH 09/10] Updated README --- README.rst | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/README.rst b/README.rst index c653a9a..1a21b8b 100644 --- a/README.rst +++ b/README.rst @@ -91,9 +91,6 @@ All configuration settings are optional. e.g: For full configuration options, see the `configuration documentation `__. - -If you are using `django-redis `__ , you can `configure `__ Django Q to use its connection pool. - Management Commands ~~~~~~~~~~~~~~~~~~~ @@ -105,6 +102,10 @@ Monitor your clusters with:: $ python manage.py qmonitor +Check overall statistics with:: + + $ python manage.py qinfo + Creating Tasks ~~~~~~~~~~~~~~ From 2efa5bac5726ab2edf76a3ea2de620d6b8de5180 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Thu, 10 Sep 2015 14:27:14 +0200 Subject: [PATCH 10/10] docs: minor adjustments --- docs/brokers.rst | 6 ++++-- docs/configure.rst | 1 + 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/docs/brokers.rst b/docs/brokers.rst index 374e2e0..396cf0b 100644 --- a/docs/brokers.rst +++ b/docs/brokers.rst @@ -1,7 +1,7 @@ Brokers ======= -The broker sits between your Django instances and your Django Q cluster instances, accepting and delivering task packages. +The broker sits between your Django instances and your Django Q cluster instances; accepting, saving and delivering task packages. Currently we support a variety of brokers from the default Redis, bleeding edge Disque to the convenient Amazon SQS. The default Redis broker does not support message receipts. @@ -37,8 +37,10 @@ The default broker for Django Q clusters. 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. +Unlike Redis, Disque supports message receipts which make delivery to the cluster workers guaranteed. +In our tests it is as fast or faster than the Redis broker. You can control the amount of time Disque should wait for completion of a task by configuring the :ref:`retry` setting. +Bulk task retrieval is supported via the :ref:`bulk` option. * Delivery receipts * Atomic diff --git a/docs/configure.rst b/docs/configure.rst index 28daf53..7f038fa 100644 --- a/docs/configure.rst +++ b/docs/configure.rst @@ -240,6 +240,7 @@ Please make sure these credentials have proper SQS access. Amazon SQS only supports a bulk setting between 1 and 10, with the total payload not exceeding 256kb. +.. _bulk: bulk ~~~~