Merge remote-tracking branch 'origin/master'

django_q/__init__.py

@@ -1,9 +1,9 @@
-import os
+# import os
-import sys
+# import sys
 import django
 
-myPath = os.path.dirname(os.path.abspath(__file__))
+# myPath = os.path.dirname(os.path.abspath(__file__))
-sys.path.insert(0, myPath)
+# sys.path.insert(0, myPath)
 
 VERSION = (0, 9, 2)
 

django_q/cluster.py

@@ -21,12 +21,11 @@ from django.utils.translation import ugettext_lazy as _
 from django import db
 
 # Local
-import signing
+from django_q import tasks
-import tasks
-
 from django_q.compat import range
 from django_q.conf import Conf, logger, psutil, get_ppid, error_reporter, rollbar
 from django_q.models import Task, Success, Schedule
+from django_q.signing import SignedPackage, BadSignature
 from django_q.status import Stat, Status
 from django_q.brokers import get_broker
 from django_q.signals import pre_execute
@@ -297,8 +296,8 @@ def pusher(task_queue, event, broker=None):
                 ack_id = task[0]
                 # unpack the task
                 try:
-                    task = signing.SignedPackage.loads(task[1])
+                    task = SignedPackage.loads(task[1])
-                except (TypeError, signing.BadSignature) as e:
+                except (TypeError, BadSignature) as e:
                     logger.error(e)
                     broker.fail(ack_id)
                     continue
@@ -325,12 +324,12 @@ def monitor(result_queue, broker=None):
             save_cached(task, broker)
         else:
             save_task(task, broker)
-        # acknowledge and log the result
+        # acknowledge result
+        ack_id = task.pop('ack_id', False)
+        if ack_id and (task['success'] or task.get('ack_failure', False)):
+            broker.acknowledge(ack_id)
+        # log the result
         if task['success']:
-            # acknowledge
-            ack_id = task.pop('ack_id', False)
-            if ack_id:
-                broker.acknowledge(ack_id)
             # log success
             logger.info(_("Processed [{}]").format(task['name']))
         else:
@@ -456,11 +455,11 @@ def save_cached(task, broker):
             if iter_count and len(group_list) == iter_count - 1:
                 group_args = '{}:{}:args'.format(broker.list_key, group)
                 # collate the results into a Task result
-                results = [signing.SignedPackage.loads(broker.cache.get(k))['result'] for k in group_list]
+                results = [SignedPackage.loads(broker.cache.get(k))['result'] for k in group_list]
                 results.append(task['result'])
                 task['result'] = results
                 task['id'] = group
-                task['args'] = signing.SignedPackage.loads(broker.cache.get(group_args))
+                task['args'] = SignedPackage.loads(broker.cache.get(group_args))
                 task.pop('iter_count', None)
                 task.pop('group', None)
                 if task.get('iter_cached', None):
@@ -479,7 +478,7 @@ def save_cached(task, broker):
                 tasks.async_chain(task['chain'], group=group, cached=task['cached'], sync=task['sync'], broker=broker)
         # save the task
         broker.cache.set(task_key,
-                         signing.SignedPackage.dumps(task),
+                         SignedPackage.dumps(task),
                          timeout)
     except Exception as e:
         logger.error(e)

django_q/conf.py

@@ -110,6 +110,11 @@ class Conf(object):
     # Number of seconds to wait for a worker to finish.
     TIMEOUT = conf.get('timeout', None)
 
+    # Whether to acknowledge unsuccessful tasks.
+    # This causes failed tasks to be considered delivered, thereby removing them from
+    # the task queue. Defaults to False.
+    ACK_FAILURES = conf.get('ack_failures', False)
+
     # 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)

django_q/status.py

@@ -2,7 +2,7 @@ import socket
 from django.utils import timezone
 from django_q.brokers import get_broker
 from django_q.conf import Conf, logger
-import signing
+from django_q.signing import SignedPackage, BadSignature
 
 
 class Status(object):
@@ -64,7 +64,7 @@ class Stat(Status):
 
     def save(self):
         try:
-            self.broker.set_stat(self.key, signing.SignedPackage.dumps(self, True), 3)
+            self.broker.set_stat(self.key, SignedPackage.dumps(self, True), 3)
         except Exception as e:
             logger.error(e)
 
@@ -83,8 +83,8 @@ class Stat(Status):
         pack = broker.get_stat(Stat.get_key(cluster_id))
         if pack:
             try:
-                return signing.SignedPackage.loads(pack)
+                return SignedPackage.loads(pack)
-            except signing.BadSignature:
+            except BadSignature:
                 return None
         return Status(cluster_id)
 
@@ -101,8 +101,8 @@ class Stat(Status):
         packs = broker.get_stats('{}:*'.format(Conf.Q_STAT)) or []
         for pack in packs:
             try:
-                stats.append(signing.SignedPackage.loads(pack))
+                stats.append(SignedPackage.loads(pack))
-            except signing.BadSignature:
+            except BadSignature:
                 continue
         return stats
 

django_q/tasks.py

@@ -1,4 +1,6 @@
 """Provides task functionality."""
+# Standard
+from time import sleep, time
 from multiprocessing import Value
 
 # django
@@ -6,9 +8,7 @@ from django.db import IntegrityError
 from django.utils import timezone
 
 # local
-import time
+from django_q.signing import SignedPackage
-import signing
-import cluster
 from django_q.conf import Conf, logger
 from django_q.models import Schedule, Task
 from django_q.humanhash import uuid
@@ -20,7 +20,7 @@ from django_q.queues import Queue
 def async(func, *args, **kwargs):
     """Queue a task for the cluster."""
     keywords = kwargs.copy()
-    opt_keys = ('hook', 'group', 'save', 'sync', 'cached', 'iter_count', 'iter_cached', 'chain', 'broker')
+    opt_keys = ('hook', 'group', 'save', 'sync', 'cached', 'ack_failure', 'iter_count', 'iter_cached', 'chain', 'broker')
     q_options = keywords.pop('q_options', {})
     # get an id
     tag = uuid()
@@ -42,13 +42,15 @@ def async(func, *args, **kwargs):
         task['cached'] = Conf.CACHED
     if 'sync' not in task and Conf.SYNC:
         task['sync'] = Conf.SYNC
+    if 'ack_failure' not in task and Conf.ACK_FAILURES:
+        task['ack_failure'] = Conf.ACK_FAILURES
     # finalize
     task['kwargs'] = keywords
     task['started'] = timezone.now()
     # signal it
     pre_enqueue.send(sender="django_q", task=task)
     # sign it
-    pack = signing.SignedPackage.dumps(task)
+    pack = SignedPackage.dumps(task)
     if task.get('sync', False):
         return _sync(pack)
     # push it
@@ -112,14 +114,14 @@ def result(task_id, wait=0, cached=Conf.CACHED):
     """
     if cached:
         return result_cached(task_id, wait)
-    start = time.time()
+    start = time()
     while True:
         r = Task.get_result(task_id)
         if r:
             return r
-        if (time.time() - start) * 1000 >= wait >= 0:
+        if (time() - start) * 1000 >= wait >= 0:
             break
-        time.sleep(0.01)
+        sleep(0.01)
 
 
 def result_cached(task_id, wait=0, broker=None):
@@ -128,14 +130,14 @@ def result_cached(task_id, wait=0, broker=None):
     """
     if not broker:
         broker = get_broker()
-    start = time.time()
+    start = time()
     while True:
         r = broker.cache.get('{}:{}'.format(broker.list_key, task_id))
         if r:
-            return signing.SignedPackage.loads(r)['result']
+            return SignedPackage.loads(r)['result']
-        if (time.time() - start) * 1000 >= wait >= 0:
+        if (time() - start) * 1000 >= wait >= 0:
             break
-        time.sleep(0.01)
+        sleep(0.01)
 
 
 def result_group(group_id, failures=False, wait=0, count=None, cached=Conf.CACHED):
@@ -150,19 +152,19 @@ def result_group(group_id, failures=False, wait=0, count=None, cached=Conf.CACHE
     """
     if cached:
         return result_group_cached(group_id, failures, wait, count)
-    start = time.time()
+    start = time()
     if count:
         while True:
-            if count_group(group_id) == count or wait and (time.time() - start) * 1000 >= wait >= 0:
+            if count_group(group_id) == count or wait and (time() - start) * 1000 >= wait >= 0:
                 break
-            time.sleep(0.01)
+            sleep(0.01)
     while True:
         r = Task.get_result_group(group_id, failures)
         if r:
             return r
-        if (time.time() - start) * 1000 >= wait >= 0:
+        if (time() - start) * 1000 >= wait >= 0:
             break
-        time.sleep(0.01)
+        sleep(0.01)
 
 
 def result_group_cached(group_id, failures=False, wait=0, count=None, broker=None):
@@ -171,24 +173,24 @@ def result_group_cached(group_id, failures=False, wait=0, count=None, broker=Non
     """
     if not broker:
         broker = get_broker()
-    start = time.time()
+    start = time()
     if count:
         while True:
-            if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait > 0:
+            if count_group_cached(group_id) == count or wait and (time() - start) * 1000 >= wait > 0:
                 break
-            time.sleep(0.01)
+            sleep(0.01)
     while True:
         group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id))
         if group_list:
             result_list = []
             for task_key in group_list:
-                task = signing.SignedPackage.loads(broker.cache.get(task_key))
+                task = SignedPackage.loads(broker.cache.get(task_key))
                 if task['success'] or failures:
                     result_list.append(task['result'])
             return result_list
-        if (time.time() - start) * 1000 >= wait >= 0:
+        if (time() - start) * 1000 >= wait >= 0:
             break
-        time.sleep(0.01)
+        sleep(0.01)
 
 
 def fetch(task_id, wait=0, cached=Conf.CACHED):
@@ -205,14 +207,14 @@ def fetch(task_id, wait=0, cached=Conf.CACHED):
     """
     if cached:
         return fetch_cached(task_id, wait)
-    start = time.time()
+    start = time()
     while True:
         t = Task.get_task(task_id)
         if t:
             return t
-        if (time.time() - start) * 1000 >= wait >= 0:
+        if (time() - start) * 1000 >= wait >= 0:
             break
-        time.sleep(0.01)
+        sleep(0.01)
 
 
 def fetch_cached(task_id, wait=0, broker=None):
@@ -221,11 +223,11 @@ def fetch_cached(task_id, wait=0, broker=None):
     """
     if not broker:
         broker = get_broker()
-    start = time.time()
+    start = time()
     while True:
         r = broker.cache.get('{}:{}'.format(broker.list_key, task_id))
         if r:
-            task = signing.SignedPackage.loads(r)
+            task = SignedPackage.loads(r)
             t = Task(id=task['id'],
                      name=task['name'],
                      func=task['func'],
@@ -237,9 +239,9 @@ def fetch_cached(task_id, wait=0, broker=None):
                      result=task['result'],
                      success=task['success'])
             return t
-        if (time.time() - start) * 1000 >= wait >= 0:
+        if (time() - start) * 1000 >= wait >= 0:
             break
-        time.sleep(0.01)
+        sleep(0.01)
 
 
 def fetch_group(group_id, failures=True, wait=0, count=None, cached=Conf.CACHED):
@@ -253,19 +255,19 @@ def fetch_group(group_id, failures=True, wait=0, count=None, cached=Conf.CACHED)
     """
     if cached:
         return fetch_group_cached(group_id, failures, wait, count)
-    start = time.time()
+    start = time()
     if count:
         while True:
-            if count_group(group_id) == count or wait and (time.time() - start) * 1000 >= wait >= 0:
+            if count_group(group_id) == count or wait and (time() - start) * 1000 >= wait >= 0:
                 break
-            time.sleep(0.01)
+            sleep(0.01)
     while True:
         r = Task.get_task_group(group_id, failures)
         if r:
             return r
-        if (time.time() - start) * 1000 >= wait >= 0:
+        if (time() - start) * 1000 >= wait >= 0:
             break
-        time.sleep(0.01)
+        sleep(0.01)
 
 
 def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None):
@@ -274,18 +276,18 @@ def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None)
     """
     if not broker:
         broker = get_broker()
-    start = time.time()
+    start = time()
     if count:
         while True:
-            if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait >= 0:
+            if count_group_cached(group_id) == count or wait and (time() - start) * 1000 >= wait >= 0:
                 break
-            time.sleep(0.01)
+            sleep(0.01)
     while True:
         group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id))
         if group_list:
             task_list = []
             for task_key in group_list:
-                task = signing.SignedPackage.loads(broker.cache.get(task_key))
+                task = SignedPackage.loads(broker.cache.get(task_key))
                 if task['success'] or failures:
                     t = Task(id=task['id'],
                              name=task['name'],
@@ -300,9 +302,9 @@ def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None)
                              success=task['success'])
                     task_list.append(t)
             return task_list
-        if (time.time() - start) * 1000 >= wait >= 0:
+        if (time() - start) * 1000 >= wait >= 0:
             break
-        time.sleep(0.01)
+        sleep(0.01)
 
 
 def count_group(group_id, failures=False, cached=Conf.CACHED):
@@ -332,7 +334,7 @@ def count_group_cached(group_id, failures=False, broker=None):
             return len(group_list)
         failure_count = 0
         for task_key in group_list:
-            task = signing.SignedPackage.loads(broker.cache.get(task_key))
+            task = SignedPackage.loads(broker.cache.get(task_key))
             if not task['success']:
                 failure_count += 1
         return failure_count
@@ -405,7 +407,7 @@ def async_iter(func, args_iter, **kwargs):
     options['cached'] = True
     # save the original arguments
     broker = options['broker']
-    broker.cache.set('{}:{}:args'.format(broker.list_key, iter_group), signing.SignedPackage.dumps(args_iter))
+    broker.cache.set('{}:{}:args'.format(broker.list_key, iter_group), SignedPackage.dumps(args_iter))
     for args in args_iter:
         if type(args) is not tuple:
             args = (args,)
@@ -671,13 +673,17 @@ class Async(object):
 
 
 def _sync(pack):
+    # Python 2.6 is unable to handle this import on top of the file
+    # because it creates a circular dependency between tasks and cluster
+    from django_q.cluster import worker, monitor
+
     """Simulate a package travelling through the cluster."""
     task_queue = Queue()
     result_queue = Queue()
-    task = signing.SignedPackage.loads(pack)
+    task = SignedPackage.loads(pack)
     task_queue.put(task)
     task_queue.put('STOP')
-    cluster.worker(task_queue, result_queue, Value('f', -1))
+    worker(task_queue, result_queue, Value('f', -1))
     result_queue.put('STOP')
-    cluster.monitor(result_queue)
+    monitor(result_queue)
     return task['id']

django_q/tests/test_cluster.py

@@ -17,7 +17,7 @@ from django_q.tasks import fetch, fetch_group, async, result, result_group, coun
 from django_q.models import Task, Success
 from django_q.conf import Conf
 from django_q.status import Stat
-from django_q.brokers import get_broker
+from django_q.brokers import get_broker, Broker
 from django_q.tests.tasks import multiply
 from django_q.queues import Queue
 
@@ -379,6 +379,57 @@ def test_update_failed(broker):
     assert saved_task.success is True
     assert saved_task.result == 'result'
 
+@pytest.mark.django_db
+def test_acknowledge_failure_override():
+    class VerifyAckMockBroker(Broker):
+        def __init__(self, *args, **kwargs):
+            super(VerifyAckMockBroker, self).__init__(*args, **kwargs)
+            self.acknowledgements = {}
+
+        def acknowledge(self, task_id):
+            count = self.acknowledgements.get(task_id, 0)
+            self.acknowledgements[task_id] = count + 1
+
+    tag = uuid()
+    task_fail_ack = {'id': tag[1],
+                     'name': tag[0],
+                     'ack_id': 'test_fail_ack_id',
+                     'ack_failure': True,
+                     'func': 'math.copysign',
+                     'args': (1, -1),
+                     'kwargs': {},
+                     'started': timezone.now(),
+                     'stopped': timezone.now(),
+                     'success': False,
+                     'result': None}
+
+    tag = uuid()
+    task_fail_no_ack = task_fail_ack.copy()
+    task_fail_no_ack.update({'id': tag[1],
+                             'name': tag[0],
+                             'ack_id': 'test_fail_no_ack_id'})
+    del task_fail_no_ack['ack_failure']
+
+    tag = uuid()
+    task_success_ack = task_fail_ack.copy()
+    task_success_ack.update({'id': tag[1],
+                             'name': tag[0],
+                             'ack_id': 'test_success_ack_id',
+                             'success': True,})
+    del task_success_ack['ack_failure']
+
+    result_queue = Queue()
+    result_queue.put(task_fail_ack)
+    result_queue.put(task_fail_no_ack)
+    result_queue.put(task_success_ack)
+    result_queue.put('STOP')
+    broker = VerifyAckMockBroker(list_key='key')
+
+    monitor(result_queue, broker)
+
+    assert broker.acknowledgements.get('test_fail_ack_id') == 1
+    assert broker.acknowledgements.get('test_fail_no_ack_id') is None
+    assert broker.acknowledgements.get('test_success_ack_id') == 1
 
 @pytest.mark.django_db
 def assert_result(task):

docs/architecture.rst

@@ -20,9 +20,13 @@ Broker
 The broker collects task packages from the django instances and queues them for pick up by a cluster.
 If the broker supports message receipts, it will keep a copy of the tasks around until a cluster acknowledges the processing of the task.
 Otherwise it is put back in the queue after a timeout period. This ensure at-least-once delivery.
-Note that even if the task errors when processed by the cluster, this is considered a successful delivery.
 Most failed deliveries will be the result of a worker or the cluster crashing before the task was saved.
 
+.. note::
+   When the :ref:`ack_failures` option is set to ``False`` (the default), a task is
+   considered a failed delivery when it raises an ``Exception``. Set
+   this option to ``True`` to acknowledge failed tasks as successful.
+
 Pusher
 """"""
 

docs/configure.rst

@@ -59,6 +59,13 @@ 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.
 
+.. _ack_failures:
+
+ack_failures
+~~~~~~~~~~~~
+
+When set to ``True``, also acknowledge unsuccessful tasks. This causes failed tasks to be considered as successful deliveries, thereby removing them from the task queue. Can also be set per-task by passing the ``ack_failure`` option to :func:`async`. Defaults to ``False``.
+
 .. _retry:
 
 retry

docs/tasks.rst

@@ -54,6 +54,10 @@ timeout
 """""""
 Overrides the cluster's timeout setting for this task.
 
+ack_failure
+"""""""""""
+Overrides the cluster's :ref:`ack_failures` setting for this task.
+
 sync
 """"
 Simulates a task execution synchronously. Useful for testing.
@@ -244,6 +248,7 @@ Reference
    :param str group: An optional group identifier
    :param int timeout: Overrides global cluster :ref:`timeout`.
    :param bool save: Overrides global save setting for this task.
+   :param bool ack_failure: Overrides the global :ref:`ack_failures` setting for this task.
    :param bool sync: If set to True, async will simulate a task execution
    :param cached: Output the result to the cache backend. Bool or timeout in seconds
    :param broker: Optional broker connection from :func:`brokers.get_broker`