From 2e52a0458475425d70b045f50939e038501e40f1 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Mon, 18 Jan 2016 10:51:04 +0100 Subject: [PATCH 1/2] Updates packages --- requirements.txt | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/requirements.txt b/requirements.txt index d03ae2a..ed6aa46 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,20 +7,21 @@ arrow==0.7.0 blessed==1.14.1 boto3==1.2.3 -botocore==1.3.17 # via boto3 +botocore==1.3.20 # via boto3 django-picklefield==0.3.2 django-redis==4.3.0 docutils==0.12 # via botocore future==0.15.2 +futures==3.0.4 # via boto3 hiredis==0.2.0 iron-core==1.2.0 # via iron-mq iron-mq==0.8 jmespath==0.9.0 # via boto3, botocore -psutil==3.3.0 +psutil==3.4.1 pymongo==3.2 python-dateutil==2.4.2 # via arrow, botocore, iron-core redis==2.10.5 requests==2.9.1 # via iron-core, rollbar rollbar==0.11.1 six==1.10.0 # via blessed, python-dateutil, rollbar -wcwidth==0.1.5 # via blessed +wcwidth==0.1.6 # via blessed From 0e8ea25396bfe2dc2506aff0f4b546fc6364efe7 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Mon, 18 Jan 2016 10:51:28 +0100 Subject: [PATCH 2/2] Only acks task if it's succesful --- django_q/cluster.py | 24 ++++++++++++++---------- 1 file changed, 14 insertions(+), 10 deletions(-) diff --git a/django_q/cluster.py b/django_q/cluster.py index 9bd4ef0..002efef 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -1,8 +1,9 @@ # Future -from __future__ import unicode_literals -from __future__ import print_function -from __future__ import division from __future__ import absolute_import +from __future__ import division +from __future__ import print_function +from __future__ import unicode_literals + from builtins import range from future import standard_library @@ -320,19 +321,21 @@ def monitor(result_queue, broker=None): name = current_process().name logger.info(_("{} monitoring at {}").format(name, current_process().pid)) for task in iter(result_queue.get, 'STOP'): - # acknowledge - ack_id = task.pop('ack_id', False) - if ack_id: - broker.acknowledge(ack_id) # save the result if task.get('cached', False): save_cached(task, broker) else: save_task(task, broker) - # log the result + # acknowledge and 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: + # log failure logger.error(_("Failed [{}] - {}").format(task['name'], task['result'])) logger.info(_("{} stopped monitoring results").format(name)) @@ -521,10 +524,11 @@ def scheduler(broker=None): # log it if not s.task: logger.error( - _('{} failed to create a task from schedule [{}]').format(current_process().name, s.name or s.id)) + _('{} failed to create a task from schedule [{}]').format(current_process().name, + s.name or s.id)) else: logger.info( - _('{} created a task from schedule [{}]').format(current_process().name, s.name or s.id)) + _('{} created a task from schedule [{}]').format(current_process().name, s.name or s.id)) # default behavior is to delete a ONCE schedule if s.schedule_type == s.ONCE: if s.repeats < 0: