From 03abbc960f8c35d0c4206c60ad01f08085539609 Mon Sep 17 00:00:00 2001 From: Balletie Date: Fri, 9 Mar 2018 21:48:00 +0100 Subject: [PATCH 1/5] Add option for acknowledging failed tasks (globally and per-task) If a task fails with an exception, it is retried until it succeeds. This is contrary to what is said in the documentation: under the "Architecture" section, heading "Broker" it says that even when a task errors, it's still considered a successful delivery. Failed tasks never get acknowledged however, thereby being retried after the timeout period. See also issues #238 and #194. This patch adds an option to acknowledge failures, thereby closing issue #238. Issue #194 would require some more work. The default of this option is set to `False`, thereby maintaining backwards compatibility. --- django_q/cluster.py | 10 +++---- django_q/conf.py | 5 ++++ django_q/tasks.py | 4 ++- django_q/tests/test_cluster.py | 53 +++++++++++++++++++++++++++++++++- docs/architecture.rst | 6 +++- docs/configure.rst | 7 +++++ docs/tasks.rst | 5 ++++ 7 files changed, 82 insertions(+), 8 deletions(-) diff --git a/django_q/cluster.py b/django_q/cluster.py index 4dd61f6..5493e06 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -325,12 +325,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: diff --git a/django_q/conf.py b/django_q/conf.py index e8afac1..2fd2e6e 100644 --- a/django_q/conf.py +++ b/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) diff --git a/django_q/tasks.py b/django_q/tasks.py index e64bfc9..1b044e8 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -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,6 +42,8 @@ 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() diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 15461d0..6c4b8c3 100644 --- a/django_q/tests/test_cluster.py +++ b/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): diff --git a/docs/architecture.rst b/docs/architecture.rst index 5fab41a..062480a 100644 --- a/docs/architecture.rst +++ b/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 """""" diff --git a/docs/configure.rst b/docs/configure.rst index 9dddc1e..638de05 100644 --- a/docs/configure.rst +++ b/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 diff --git a/docs/tasks.rst b/docs/tasks.rst index 7b3160d..230d916 100644 --- a/docs/tasks.rst +++ b/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` From ae9392137c66832e2e4fa0a51938aad2e6fdb8a4 Mon Sep 17 00:00:00 2001 From: Eagllus Date: Tue, 13 Mar 2018 09:21:01 +0100 Subject: [PATCH 2/5] Change path location of django q --- django_q/__init__.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/django_q/__init__.py b/django_q/__init__.py index 092c838..5ccb120 100644 --- a/django_q/__init__.py +++ b/django_q/__init__.py @@ -1,9 +1,9 @@ -import os -import sys +# import os +# import sys import django -myPath = os.path.dirname(os.path.abspath(__file__)) -sys.path.insert(0, myPath) +# myPath = os.path.dirname(os.path.abspath(__file__)) +# sys.path.insert(0, myPath) VERSION = (0, 9, 2) From 9c395464173dd8cf0afcb42c09265df824c1039d Mon Sep 17 00:00:00 2001 From: Eagllus Date: Tue, 13 Mar 2018 09:22:09 +0100 Subject: [PATCH 3/5] Change imports signing classes explicit --- django_q/cluster.py | 12 ++++++------ django_q/status.py | 12 ++++++------ django_q/tasks.py | 18 +++++++++--------- 3 files changed, 21 insertions(+), 21 deletions(-) diff --git a/django_q/cluster.py b/django_q/cluster.py index 4dd61f6..d62c846 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -21,12 +21,12 @@ from django.utils.translation import ugettext_lazy as _ from django import db # Local -import signing 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 +297,8 @@ def pusher(task_queue, event, broker=None): ack_id = task[0] # unpack the task try: - task = signing.SignedPackage.loads(task[1]) - except (TypeError, signing.BadSignature) as e: + task = SignedPackage.loads(task[1]) + except (TypeError, BadSignature) as e: logger.error(e) broker.fail(ack_id) continue @@ -456,11 +456,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 +479,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) diff --git a/django_q/status.py b/django_q/status.py index db03825..51f32ee 100644 --- a/django_q/status.py +++ b/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) - except signing.BadSignature: + return SignedPackage.loads(pack) + 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)) - except signing.BadSignature: + stats.append(SignedPackage.loads(pack)) + except BadSignature: continue return stats diff --git a/django_q/tasks.py b/django_q/tasks.py index e64bfc9..e64600b 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -7,7 +7,7 @@ from django.utils import timezone # local import time -import signing +from django_q.signing import SignedPackage import cluster from django_q.conf import Conf, logger from django_q.models import Schedule, Task @@ -48,7 +48,7 @@ def async(func, *args, **kwargs): # 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 @@ -132,7 +132,7 @@ def result_cached(task_id, wait=0, broker=None): 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: break time.sleep(0.01) @@ -182,7 +182,7 @@ def result_group_cached(group_id, failures=False, wait=0, count=None, broker=Non 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 @@ -225,7 +225,7 @@ def fetch_cached(task_id, wait=0, broker=None): 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'], @@ -285,7 +285,7 @@ def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None) 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'], @@ -332,7 +332,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 +405,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,) @@ -674,7 +674,7 @@ def _sync(pack): """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)) From 8f87bc352f081284e4939a7eaa2a9a45ccc726d1 Mon Sep 17 00:00:00 2001 From: Eagllus Date: Tue, 13 Mar 2018 09:28:17 +0100 Subject: [PATCH 4/5] Change imports for cluster and tasks --- django_q/cluster.py | 3 +- django_q/tasks.py | 69 +++++++++++++++++++++++---------------------- 2 files changed, 36 insertions(+), 36 deletions(-) diff --git a/django_q/cluster.py b/django_q/cluster.py index d62c846..403b649 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -21,8 +21,7 @@ from django.utils.translation import ugettext_lazy as _ from django import db # Local -import tasks - +from django_q 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 diff --git a/django_q/tasks.py b/django_q/tasks.py index e64600b..96a43fd 100644 --- a/django_q/tasks.py +++ b/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,8 @@ from django.db import IntegrityError from django.utils import timezone # local -import time from django_q.signing import SignedPackage -import cluster +from django_q import cluster from django_q.conf import Conf, logger from django_q.models import Schedule, Task from django_q.humanhash import uuid @@ -112,14 +113,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 +129,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 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 +151,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,12 +172,12 @@ 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: @@ -186,9 +187,9 @@ def result_group_cached(group_id, failures=False, wait=0, count=None, broker=Non 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 +206,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,7 +222,7 @@ 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: @@ -237,9 +238,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 +254,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,12 +275,12 @@ 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: @@ -300,9 +301,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): From 430c177b9241d654847481a0641ff973bfe6da3c Mon Sep 17 00:00:00 2001 From: Eagllus Date: Tue, 13 Mar 2018 09:37:13 +0100 Subject: [PATCH 5/5] Move import to the function that uses it Only Python 2.x is having problems with this import so for now the fix would be to move the import to the function that uses cluster --- django_q/tasks.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/django_q/tasks.py b/django_q/tasks.py index 96a43fd..bf94a2a 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -9,7 +9,6 @@ from django.utils import timezone # local from django_q.signing import SignedPackage -from django_q import cluster from django_q.conf import Conf, logger from django_q.models import Schedule, Task from django_q.humanhash import uuid @@ -672,13 +671,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 = 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']