From c350ea13f1dabcf9a3224049ff780f68eec6becd Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Fri, 24 Jul 2015 14:57:33 +0200 Subject: [PATCH 1/5] monitor TQ will indicate green when `queue_limit` has been reached. --- django_q/monitor.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/django_q/monitor.py b/django_q/monitor.py index d575e11..92bd7ff 100644 --- a/django_q/monitor.py +++ b/django_q/monitor.py @@ -53,9 +53,11 @@ def monitor(run_once=False): elif stat.status == Conf.IDLE: status = str(Conf.IDLE) # color q's - tasks = stat.task_q_size - if tasks > 0: - tasks = term.cyan(str(tasks)) + tasks = str(stat.task_q_size) + if stat.task_q_size > 0: + tasks = term.cyan(str(stat.task_q_size)) + if Conf.QUEUE_LIMIT and stat.task_q_size == Conf.QUEUE_LIMIT: + tasks = term.green(str(stat.task_q_size)) results = stat.done_q_size if results > 0: results = term.cyan(str(results)) From 38829f25c2411cc48e17077ce29ea6e69dbbfdf1 Mon Sep 17 00:00:00 2001 From: Ilan Date: Fri, 24 Jul 2015 20:47:35 +0200 Subject: [PATCH 2/5] adds group filter to admin --- django_q/admin.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/django_q/admin.py b/django_q/admin.py index 2f80ade..3fbfdce 100644 --- a/django_q/admin.py +++ b/django_q/admin.py @@ -26,6 +26,7 @@ class TaskAdmin(admin.ModelAdmin): search_fields = ('name', 'func', 'group') readonly_fields = [] + list_filter = ('group',) def get_readonly_fields(self, request, obj=None): return list(self.readonly_fields) + \ @@ -56,6 +57,7 @@ class FailAdmin(admin.ModelAdmin): actions = [retry_failed] search_fields = ('name', 'func') + list_filter = ('group',) readonly_fields = [] def get_readonly_fields(self, request, obj=None): From 25b4b03544db7e8438ee0043814525d085cad438 Mon Sep 17 00:00:00 2001 From: Ilan Date: Sat, 25 Jul 2015 16:58:48 +0200 Subject: [PATCH 3/5] Updated future==0.15.0 --- requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index a7c1cfa..83e7e99 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,7 +8,7 @@ arrow==0.6.0 blessed==1.9.5 django-picklefield==0.3.1 django-redis==4.2.0 -future==0.14.3 +future==0.15.0 hiredis==0.2.0 msgpack-python==0.4.6 # via django-redis psutil==3.1.1 From 82bd93d67f56279023214189e556e2f32b0f0002 Mon Sep 17 00:00:00 2001 From: Ilan Date: Sat, 25 Jul 2015 17:01:06 +0200 Subject: [PATCH 4/5] docstring linting --- django_q/admin.py | 25 +++++++++++++++++++------ 1 file changed, 19 insertions(+), 6 deletions(-) diff --git a/django_q/admin.py b/django_q/admin.py index 3fbfdce..f03e90f 100644 --- a/django_q/admin.py +++ b/django_q/admin.py @@ -1,3 +1,4 @@ +"""Admin module for Django.""" from django.contrib import admin from django.utils.translation import ugettext_lazy as _ @@ -6,6 +7,9 @@ from .models import Success, Failure, Schedule class TaskAdmin(admin.ModelAdmin): + + """model admin for success tasks.""" + list_display = ( u'name', 'func', @@ -16,11 +20,11 @@ class TaskAdmin(admin.ModelAdmin): ) def has_add_permission(self, request, obj=None): - """Don't allow adds""" + """Don't allow adds.""" return False def get_queryset(self, request): - """Only show successes""" + """Only show successes.""" qs = super(TaskAdmin, self).get_queryset(request) return qs.filter(success=True) @@ -29,11 +33,13 @@ class TaskAdmin(admin.ModelAdmin): list_filter = ('group',) def get_readonly_fields(self, request, obj=None): - return list(self.readonly_fields) + \ - [field.name for field in obj._meta.fields] + """Set all fields readonly.""" + return list(self.readonly_fields) +\ + [field.name for field in obj._meta.fields] def retry_failed(FailAdmin, request, queryset): + """Submit selected tasks back to the queue.""" for task in queryset: async(task.func, *task.args or (), hook=task.hook, **task.kwargs or {}) task.delete() @@ -43,6 +49,9 @@ retry_failed.short_description = _("Resubmit selected tasks to queue") class FailAdmin(admin.ModelAdmin): + + """model admin for failed tasks.""" + list_display = ( 'name', 'func', @@ -52,7 +61,7 @@ class FailAdmin(admin.ModelAdmin): ) def has_add_permission(self, request, obj=None): - """Don't allow adds""" + """Don't allow adds.""" return False actions = [retry_failed] @@ -61,11 +70,15 @@ class FailAdmin(admin.ModelAdmin): readonly_fields = [] def get_readonly_fields(self, request, obj=None): + """Set all fields readonly.""" return list(self.readonly_fields) + \ - [field.name for field in obj._meta.fields] + [field.name for field in obj._meta.fields] class ScheduleAdmin(admin.ModelAdmin): + + """ model admin for schedules """ + list_display = ( 'id', 'name', From ba58b4face26464c519107d9851ca2369a923974 Mon Sep 17 00:00:00 2001 From: Ilan Date: Mon, 27 Jul 2015 12:31:38 +0200 Subject: [PATCH 5/5] closes db connection on worker and monitor spawn This should prevent some problems with Postgresql where the workers would use stale db connections and cause errors. --- django_q/cluster.py | 122 +++++++++++++++++++++++--------------------- 1 file changed, 64 insertions(+), 58 deletions(-) diff --git a/django_q/cluster.py b/django_q/cluster.py index 00e1929..e63ed30 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -31,6 +31,7 @@ except ImportError: # Django from django.utils import timezone from django.utils.translation import ugettext_lazy as _ +from django import db # Local import signing @@ -328,6 +329,7 @@ def monitor(result_queue): """ 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'): save_task(task) if task['success']: @@ -346,6 +348,7 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT): """ name = current_process().name logger.info(_('{} ready for work at {}').format(name, current_process().pid)) + db.close_old_connections() task_count = 0 # Start reading the task queue for pack in iter(task_queue.get, 'STOP'): @@ -399,9 +402,9 @@ def save_task(task): if not task.get('save', Conf.SAVE_LIMIT > 0) and task['success']: return # SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning - if task['success'] and 0 < Conf.SAVE_LIMIT < Success.objects.count(): - Success.objects.last().delete() try: + if task['success'] and 0 < Conf.SAVE_LIMIT < Success.objects.count(): + Success.objects.last().delete() Task.objects.create(id=task['id'], name=task['name'], func=task['func'], @@ -421,62 +424,65 @@ def scheduler(list_key=Conf.Q_LIST): """ Creates a task from a schedule at the scheduled time and schedules next run """ - for s in Schedule.objects.exclude(repeats=0).filter(next_run__lt=timezone.now()): - args = () - kwargs = {} - # get args, kwargs and hook - if s.kwargs: - try: - # eval should be safe here cause dict() - kwargs = eval('dict({})'.format(s.kwargs)) - except SyntaxError: - kwargs = {} - if s.args: - args = ast.literal_eval(s.args) - # single value won't eval to tuple, so: - if type(args) != tuple: - args = (args,) - q_options = kwargs.get('q_options', {}) - if s.hook: - q_options['hook'] = s.hook - # set up the next run time - if not s.schedule_type == s.ONCE: - next_run = arrow.get(s.next_run) - if s.schedule_type == s.HOURLY: - next_run = next_run.replace(hours=+1) - elif s.schedule_type == s.DAILY: - next_run = next_run.replace(days=+1) - elif s.schedule_type == s.WEEKLY: - next_run = next_run.replace(weeks=+1) - elif s.schedule_type == s.MONTHLY: - next_run = next_run.replace(months=+1) - elif s.schedule_type == s.QUARTERLY: - next_run = next_run.replace(months=+3) - elif s.schedule_type == s.YEARLY: - next_run = next_run.replace(years=+1) - s.next_run = next_run.datetime - s.repeats += -1 - # send it to the cluster - q_options['list_key'] = list_key - q_options['group'] = s.name or s.id - kwargs['q_options'] = q_options - s.task = tasks.async(s.func, *args, **kwargs) - # log it - if not s.task: - logger.error( - _('{} 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)) - # default behavior is to delete a ONCE schedule - if s.schedule_type == s.ONCE: - if s.repeats < 0: - s.delete() - return - # but not if it has a positive repeats - s.repeats = 0 - # save the schedule - s.save() + try: + for s in Schedule.objects.exclude(repeats=0).filter(next_run__lt=timezone.now()): + args = () + kwargs = {} + # get args, kwargs and hook + if s.kwargs: + try: + # eval should be safe here cause dict() + kwargs = eval('dict({})'.format(s.kwargs)) + except SyntaxError: + kwargs = {} + if s.args: + args = ast.literal_eval(s.args) + # single value won't eval to tuple, so: + if type(args) != tuple: + args = (args,) + q_options = kwargs.get('q_options', {}) + if s.hook: + q_options['hook'] = s.hook + # set up the next run time + if not s.schedule_type == s.ONCE: + next_run = arrow.get(s.next_run) + if s.schedule_type == s.HOURLY: + next_run = next_run.replace(hours=+1) + elif s.schedule_type == s.DAILY: + next_run = next_run.replace(days=+1) + elif s.schedule_type == s.WEEKLY: + next_run = next_run.replace(weeks=+1) + elif s.schedule_type == s.MONTHLY: + next_run = next_run.replace(months=+1) + elif s.schedule_type == s.QUARTERLY: + next_run = next_run.replace(months=+3) + elif s.schedule_type == s.YEARLY: + next_run = next_run.replace(years=+1) + s.next_run = next_run.datetime + s.repeats += -1 + # send it to the cluster + q_options['list_key'] = list_key + q_options['group'] = s.name or s.id + kwargs['q_options'] = q_options + s.task = tasks.async(s.func, *args, **kwargs) + # log it + if not s.task: + logger.error( + _('{} 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)) + # default behavior is to delete a ONCE schedule + if s.schedule_type == s.ONCE: + if s.repeats < 0: + s.delete() + return + # but not if it has a positive repeats + s.repeats = 0 + # save the schedule + s.save() + except Exception as e: + logger.error(e) def set_cpu_affinity(n, process_ids, actual=not Conf.TESTING):