From b113a528256a1df40a64082ea6426a673b1109c8 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Fri, 3 Jul 2015 15:12:11 +0200 Subject: [PATCH] Restructured all the modules The core module was becoming too large. By splitting it in cluster, monitor and tasks it becomes more manageable. It also reflects the documentation structure. --- django_q/__init__.py | 27 +-- django_q/admin.py | 10 +- django_q/{core.py => cluster.py} | 265 +++-------------------- django_q/conf.py | 30 ++- django_q/management/commands/qcluster.py | 6 +- django_q/management/commands/qmonitor.py | 76 +------ django_q/models.py | 3 + django_q/monitor.py | 184 ++++++++++++++++ django_q/tasks.py | 119 ++++++++++ django_q/tests/test_admin.py | 4 +- django_q/tests/test_cluster.py | 25 ++- django_q/tests/test_monitor.py | 4 +- django_q/tests/test_scheduler.py | 7 +- 13 files changed, 399 insertions(+), 361 deletions(-) rename django_q/{core.py => cluster.py} (62%) create mode 100644 django_q/monitor.py create mode 100644 django_q/tasks.py diff --git a/django_q/__init__.py b/django_q/__init__.py index 2a29cb3..ee26cd6 100644 --- a/django_q/__init__.py +++ b/django_q/__init__.py @@ -1,29 +1,6 @@ -from django_q.models import Task, Schedule -from django_q.core import async, schedule +from .tasks import async, schedule, result, fetch +from .models import Task, Schedule VERSION = (0, 1, 4) default_app_config = 'django_q.apps.DjangoQConfig' - - -def result(name): - """ - Returns the result of the named task - :type name: str or unicode - :param name: the task name - :return: the result object of this task - :rtype: object or str - """ - return Task.get_result(name) - - -def get_task(name): - """ - Returns the processed task - :param name: the task name - :type name: str or unicode - :return: the full task object - :rtype: Task - """ - if Task.objects.filter(name=name).exists(): - return Task.objects.get(name=name) diff --git a/django_q/admin.py b/django_q/admin.py index 4df8370..6c1f643 100644 --- a/django_q/admin.py +++ b/django_q/admin.py @@ -1,6 +1,6 @@ from django.contrib import admin - -from django_q.core import async +from django.utils.translation import ugettext_lazy as _ +from .tasks import async from .models import Success, Failure, Schedule @@ -35,12 +35,12 @@ def retry_failed(FailAdmin, request, queryset): task.delete() -retry_failed.short_description = "Resubmit selected tasks to Q" +retry_failed.short_description = _("Resubmit selected tasks to queue") class FailAdmin(admin.ModelAdmin): list_display = ( - u'name', + 'name', 'func', 'started', 'result' @@ -61,7 +61,7 @@ class FailAdmin(admin.ModelAdmin): class ScheduleAdmin(admin.ModelAdmin): list_display = ( - u'id', + 'id', 'func', 'schedule_type', 'repeats', diff --git a/django_q/core.py b/django_q/cluster.py similarity index 62% rename from django_q/core.py rename to django_q/cluster.py index 5101eeb..b35c997 100644 --- a/django_q/core.py +++ b/django_q/cluster.py @@ -3,53 +3,39 @@ from __future__ import unicode_literals from __future__ import print_function from __future__ import division from __future__ import absolute_import -import ast from builtins import range - from future import standard_library standard_library.install_aliases() # Standard import importlib -import logging import os import signal -from multiprocessing import Queue, Event, Process, Value, current_process import socket import sys +import ast from time import sleep +from multiprocessing import Queue, Event, Process, Value, current_process try: import cPickle as pickle except ImportError: import pickle -# External -import redis +# external import arrow # Django from django.core import signing from django.utils import timezone +from django.utils.translation import ugettext_lazy as _ # Local -from .conf import Conf -from .humanhash import uuid +from .conf import Conf, redis_client, logger from .models import Task, Success, Schedule - -logger = logging.getLogger('django-q') - -# Set up standard logging handler in case there is none -if not logger.handlers: - logger.setLevel(level=getattr(logging, Conf.LOG_LEVEL)) - formatter = logging.Formatter(fmt='%(asctime)s [Q] %(levelname)s %(message)s', - datefmt='%H:%M:%S') - handler = logging.StreamHandler() - handler.setFormatter(formatter) - logger.addHandler(handler) - -redis_client = redis.StrictRedis(**Conf.REDIS) +from .monitor import Status, Stat +from .tasks import SignedPackage, async class Cluster(object): @@ -82,7 +68,7 @@ class Cluster(object): self.start_event = Event() self.sentinel = Process(target=Sentinel, args=(self.stop_event, self.start_event, self.list_key, self.timeout)) self.sentinel.start() - logger.info('Q Cluster-{} starting.'.format(self.pid)) + logger.info(_('Q Cluster-{} starting.').format(self.pid)) while not self.start_event.is_set(): sleep(0.2) return self.pid @@ -90,16 +76,16 @@ class Cluster(object): def stop(self): if not self.sentinel.is_alive(): return False - logger.info('Q Cluster-{} stopping.'.format(self.pid)) + logger.info(_('Q Cluster-{} stopping.').format(self.pid)) self.stop_event.set() self.sentinel.join() - logger.info('Q Cluster-{} has stopped.'.format(self.pid)) + logger.info(_('Q Cluster-{} has stopped.').format(self.pid)) self.start_event = None self.stop_event = None return True def sig_handler(self, signum, frame): - logger.debug('{} got signal {}'.format(current_process().name, Conf.SIGNAL_NAMES.get(signum, 'UNKNOWN'))) + logger.debug(_('{} got signal {}').format(current_process().name, Conf.SIGNAL_NAMES.get(signum, 'UNKNOWN'))) self.stop() @property @@ -201,17 +187,17 @@ class Sentinel(object): process.terminate() if process == self.monitor: self.monitor = self.spawn_monitor() - logger.error("reincarnated monitor {} after sudden.".format(process.pid)) + logger.error(_("reincarnated monitor {} after sudden.").format(process.pid)) elif process == self.pusher: self.pusher = self.spawn_pusher() - logger.error("reincarnated pusher {} after sudden death".format(process.pid)) + logger.error(_("reincarnated pusher {} after sudden death.").format(process.pid)) else: self.pool.remove(process) self.spawn_worker() if int(process.timer.value) >= self.timeout: - logger.warn("reincarnated worker {} after timeout.".format(process.pid)) + logger.warn(_("reincarnated worker {} after timeout.").format(process.pid)) else: - logger.error("reincarnated worker {} after sudden death.".format(process.pid)) + logger.error(_("reincarnated worker {} after sudden death.").format(process.pid)) self.reincarnations += 1 def spawn_cluster(self): @@ -222,10 +208,10 @@ class Sentinel(object): self.pusher = self.spawn_pusher() def guard(self): - logger.info('{} guarding cluster at {}'.format(current_process().name, self.pid)) + logger.info(_('{} guarding cluster at {}').format(current_process().name, self.pid)) self.start_event.set() Stat(self).save() - logger.info('Q Cluster-{} running.'.format(self.parent_pid)) + logger.info(_('Q Cluster-{} running.').format(self.parent_pid)) scheduler(list_key=self.list_key) counter = 0 # Guard loop. Runs at least once @@ -258,13 +244,14 @@ class Sentinel(object): def stop(self): Stat(self).save() name = current_process().name - logger.info('{} stopping pool processes'.format(name)) + logger.info('{} stopping cluster processes'.format(name)) # Stopping pusher self.event_out.set() + # Wait for it to stop while self.pusher.is_alive(): sleep(0.2) Stat(self).save() - # Putting poison pills in the queue + # Put poison pills in the queue for _ in range(self.pool_size): self.task_queue.put('STOP') # Wait for all the workers to exit @@ -290,16 +277,16 @@ def pusher(task_queue, e, list_key=Conf.Q_LIST, r=redis_client): :type e: multiprocessing.Event :type list_key: str """ - logger.info('{} pushing tasks at {}'.format(current_process().name, current_process().pid)) + logger.info(_('{} pushing tasks at {}').format(current_process().name, current_process().pid)) while True: task = r.blpop(list_key, 1) if task: task = task[1] task_queue.put(task) - logger.debug('queueing {}'.format(task)) + logger.debug(_('queueing {}').format(task)) if e.is_set(): break - logger.info("{} stopped pushing tasks".format(current_process().name)) + logger.info(_("{} stopped pushing tasks.").format(current_process().name)) def monitor(done_queue): @@ -308,14 +295,14 @@ def monitor(done_queue): :type done_queue: multiprocessing.Queue """ name = current_process().name - logger.info("{} monitoring at {}".format(name, current_process().pid)) + logger.info(_("{} monitoring at {}").format(name, current_process().pid)) for task in iter(done_queue.get, 'STOP'): if task['success']: - logger.info("Processed [{}]".format(task['name'])) + logger.info(_("Processed [{}]").format(task['name'])) else: - logger.error("Failed [{}] - {}".format(task['name'], task['result'])) + logger.error(_("Failed [{}] - {}").format(task['name'], task['result'])) save_task(task) - logger.info("{} stopped monitoring results".format(name)) + logger.info(_("{} stopped monitoring results").format(name)) def worker(task_queue, done_queue, timer): @@ -326,7 +313,7 @@ def worker(task_queue, done_queue, timer): :type timer: multiprocessing.Value """ name = current_process().name - logger.info('{} ready for work at {}'.format(name, current_process().pid)) + logger.info(_('{} ready for work at {}').format(name, current_process().pid)) task_count = 0 # Start reading the task queue for pack in iter(task_queue.get, 'STOP'): @@ -340,7 +327,7 @@ def worker(task_queue, done_queue, timer): logger.error(e) continue # Get the function from the task - logger.info('{} processing [{}]'.format(name, task['name'])) + logger.info(_('{} processing [{}]').format(name, task['name'])) f = task['func'] # if it's not an instance try to get it from the string if not callable(task['func']): @@ -368,7 +355,7 @@ def worker(task_queue, done_queue, timer): # Recycle if task_count == Conf.RECYCLE and task_queue.qsize() == 0: break - logger.info('{} stopped doing work'.format(name)) + logger.info(_('{} stopped doing work').format(name)) def save_task(task): @@ -396,192 +383,6 @@ def save_task(task): logger.exception(e) -def async(func, *args, **kwargs): - """ - Sends a task to the cluster - """ - - hook = kwargs.pop('hook', None) - list_key = kwargs.pop('list_key', Conf.Q_LIST) - r = kwargs.pop('redis', redis_client) - - task = {'name': uuid()[0], 'func': func, 'hook': hook, 'args': args, 'kwargs': kwargs, 'started': timezone.now()} - pack = SignedPackage.dumps(task) - r.rpush(list_key, pack) - logger.debug('Pushed {}'.format(pack)) - return task['name'] - - -class SignedPackage(object): - """ - Wraps Django's signing module with custom Pickle serializer - """ - - @staticmethod - def dumps(obj, compressed=Conf.COMPRESSED): - return signing.dumps(obj, - key=Conf.SECRET_KEY, - salt='django_q.q', - compress=compressed, - serializer=PickleSerializer) - - @staticmethod - def loads(obj): - return signing.loads(obj, - key=Conf.SECRET_KEY, - salt='django_q.q', - serializer=PickleSerializer) - - -class PickleSerializer(object): - """ - Simple wrapper around Pickle for signing.dumps and - signing.loads. - """ - - @staticmethod - def dumps(obj): - return pickle.dumps(obj) - - @staticmethod - def loads(data): - return pickle.loads(data) - - -class Status(object): - """ - Cluster status base object - """ - - def __init__(self, pid): - self.workers = [] - self.tob = None - self.reincarnations = 0 - self.cluster_id = pid - self.sentinel = 0 - self.status = 'Idle' - self.done_q_size = 0 - self.host = socket.gethostname() - self.monitor = 0 - self.task_q_size = 0 - self.pusher = 0 - self.timestamp = timezone.now() - - -class Stat(Status): - """ - Status object for Cluster monitoring - """ - - def __init__(self, sentinel): - super(Stat, self).__init__(sentinel.parent_pid) - self.r = sentinel.r - self.tob = sentinel.tob - self.reincarnations = sentinel.reincarnations - self.sentinel = sentinel.pid - self.status = sentinel.status() - self.done_q_size = sentinel.done_queue.qsize() - if sentinel.monitor: - self.monitor = sentinel.monitor.pid - self.task_q_size = sentinel.task_queue.qsize() - if sentinel.pusher: - self.pusher = sentinel.pusher.pid - for w in sentinel.pool: - self.workers.append(w.pid) - - def uptime(self): - return (timezone.now() - self.tob).total_seconds() - - @property - def key(self): - """ - :return: redis key for this cluster statistic - """ - return self.get_key(self.cluster_id) - - @staticmethod - def get_key(cluster_id): - """ - :param cluster_id: cluster ID - :return: redis key for the cluster statistic - """ - return '{}:{}'.format(Conf.Q_STAT, cluster_id) - - def save(self): - self.r.set(self.key, SignedPackage.dumps(self, True), 3) - - def empty_queues(self): - return self.done_q_size + self.task_q_size == 0 - - @staticmethod - def get(cluster_id, r=redis_client): - """ - gets the current status for the cluster - :param cluster_id: id of the cluster - :return: Stat or Status - """ - key = Stat.get_key(cluster_id) - if r.exists(key): - pack = r.get(key) - try: - return SignedPackage.loads(pack) - except signing.BadSignature: - return None - return Status(cluster_id) - - @staticmethod - def get_all(r=redis_client): - """ - Gets status for all currently running clusters with the same prefix and secret key - :return: Stat list - """ - stats = [] - keys = r.keys(pattern='{}:*'.format(Conf.Q_STAT)) - if keys: - packs = r.mget(keys) - for pack in packs: - try: - stats.append(SignedPackage.loads(pack)) - except signing.BadSignature: - continue - return stats - - def __getstate__(self): - # Don't pickle the redis connection - state = dict(self.__dict__) - del state['r'] - return state - - -def schedule(func, *args, **kwargs): - """ - :param func: function to schedule - :param args: function arguments - :param hook: optional result hook function - :type schedule_type: Schedule.TYPE - :param repeats: how many times to repeat. 0=never, -1=always - :param next_run: Next scheduled run - :type next_run: datetime.datetime - :param kwargs: function keyword arguments - :return: the schedule object - :rtype: Schedule - """ - - hook = kwargs.pop('hook', None) - schedule_type = kwargs.pop('schedule_type', Schedule.ONCE) - repeats = kwargs.pop('repeats', -1) - next_run = kwargs.pop('next_run', timezone.now()) - - return Schedule.objects.create(func=func, - hook=hook, - args=args, - kwargs=kwargs, - schedule_type=schedule_type, - repeats=repeats, - next_run=next_run - ) - - def scheduler(list_key=Conf.Q_LIST): """ Creates a task from a schedule at the scheduled time and schedules next run @@ -626,7 +427,7 @@ def scheduler(list_key=Conf.Q_LIST): kwargs['list_key'] = list_key s.task = async(s.func, *args, **kwargs) if not s.task: - logger.error('{} failed to create task from schedule {}').format(current_process().name, s.id) + logger.error(_('{} failed to create task from schedule {}').format(current_process().name, s.id)) else: - logger.info('{} created [{}] from schedule {}'.format(current_process().name, s.task, s.id)) - s.save() + logger.info(_('{} created [{}] from schedule {}').format(current_process().name, s.task, s.id)) + s.save() \ No newline at end of file diff --git a/django_q/conf.py b/django_q/conf.py index 26af4cf..c19fc8c 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -1,7 +1,10 @@ +import logging from signal import signal from multiprocessing import cpu_count +from django.utils.translation import ugettext_lazy as _ from django.conf import settings +import redis class Conf(object): @@ -52,9 +55,24 @@ class Conf(object): # Getting the signal names SIGNAL_NAMES = dict((getattr(signal, n), n) for n in dir(signal) if n.startswith('SIG') and '_' not in n) - # Cluster status descriptions - STARTING = 'Starting' - WORKING = 'Working' - IDLE = "Idle" - STOPPED = 'Stopped' - STOPPING = 'Stopping' + # Translators: Cluster status descriptions + STARTING = _('Starting') + WORKING = _('Working') + IDLE = _("Idle") + STOPPED = _('Stopped') + STOPPING = _('Stopping') + +# logger +logger = logging.getLogger('django-q') + +# Set up standard logging handler in case there is none +if not logger.handlers: + logger.setLevel(level=getattr(logging, Conf.LOG_LEVEL)) + formatter = logging.Formatter(fmt='%(asctime)s [Q] %(levelname)s %(message)s', + datefmt='%H:%M:%S') + handler = logging.StreamHandler() + handler.setFormatter(formatter) + logger.addHandler(handler) + +# redis client +redis_client = redis.StrictRedis(**Conf.REDIS) diff --git a/django_q/management/commands/qcluster.py b/django_q/management/commands/qcluster.py index 93742e3..a1ac799 100644 --- a/django_q/management/commands/qcluster.py +++ b/django_q/management/commands/qcluster.py @@ -1,9 +1,11 @@ from django.core.management.base import BaseCommand -from django_q.core import Cluster +from django.utils.translation import ugettext as _ +from django_q.cluster import Cluster class Command(BaseCommand): - help = "Starts a Django Q Cluster." + # Translators: help text for qcluster management command + help = _("Starts a Django Q Cluster.") def handle(self, *args, **options): q = Cluster() diff --git a/django_q/management/commands/qmonitor.py b/django_q/management/commands/qmonitor.py index ee145d3..3de605c 100644 --- a/django_q/management/commands/qmonitor.py +++ b/django_q/management/commands/qmonitor.py @@ -1,84 +1,16 @@ # Django from django.core.management.base import BaseCommand -from django.utils import timezone +from django.utils.translation import ugettext as _ -# External -from blessed import Terminal -# Local -from django_q.core import Stat, redis_client -from django_q.conf import Conf +from ...monitor import monitor -# TODO add name argument to monitor different clusters class Command(BaseCommand): - help = "Monitors cluster activity" + # Translators: help text for qmonitor management command + help = _("Monitors Q Cluster activity") def handle(self, *args, **options): monitor() -def monitor(run_once=False): - term = Terminal() - r = redis_client - with term.fullscreen(), term.hidden_cursor(), term.cbreak(): - val = None - start_width = int(term.width / 8) - while val not in (u'q', u'Q',): - col_width = int(term.width / 8) - # In case of resize - if col_width != start_width: - print(term.clear) - start_width = col_width - print(term.move(0, 0) + term.black_on_green(term.center('Host', width=col_width - 1))) - print(term.move(0, 1 * col_width) + term.black_on_green(term.center('Id', width=col_width - 1))) - print(term.move(0, 2 * col_width) + term.black_on_green(term.center('Status', width=col_width - 1))) - print(term.move(0, 3 * col_width) + term.black_on_green(term.center('Pool', width=col_width - 1))) - print(term.move(0, 4 * col_width) + term.black_on_green(term.center('TQ', width=col_width - 1))) - print(term.move(0, 5 * col_width) + term.black_on_green(term.center('RQ', width=col_width - 1))) - print(term.move(0, 6 * col_width) + term.black_on_green(term.center('RC', width=col_width - 1))) - print(term.move(0, 7 * col_width) + term.black_on_green(term.center('Up', width=col_width - 1))) - i = 2 - stats = Stat.get_all(r=r) - print(term.clear_eos()) - for stat in stats: - # color status - if stat.status == Conf.WORKING: - status = term.green(Conf.WORKING) - elif stat.status == Conf.STOPPED: - status = term.red(Conf.STOPPED) - elif stat.status == Conf.IDLE: - status = Conf.IDLE - else: - status = term.yellow(stat.status) - # color q's - tasks = stat.task_q_size - if tasks > 0: - tasks = term.cyan(str(tasks)) - results = stat.done_q_size - if results > 0: - results = term.cyan(str(results)) - # color workers - workers = len(stat.workers) - if workers < Conf.WORKERS: - workers = term.yellow(str(workers)) - # format uptime - uptime = (timezone.now() - stat.tob).total_seconds() - hours, remainder = divmod(uptime, 3600) - minutes, seconds = divmod(remainder, 60) - uptime = '%d:%02d:%02d' % (hours, minutes, seconds) - # print to the terminal - print(term.move(i, 0) + term.center(stat.host[:col_width - 1], width=col_width - 1)) - print(term.move(i, 1 * col_width) + term.center(stat.cluster_id, width=col_width - 1)) - print(term.move(i, 2 * col_width) + term.center(status, width=col_width - 1)) - print(term.move(i, 3 * col_width) + term.center(workers, width=col_width - 1)) - print(term.move(i, 4 * col_width) + term.center(tasks, width=col_width - 1)) - print(term.move(i, 5 * col_width) + term.center(results, width=col_width - 1)) - print(term.move(i, 6 * col_width) + term.center(stat.reincarnations, width=col_width - 1)) - print(term.move(i, 7 * col_width) + term.center(uptime, width=col_width - 1)) - i += 1 - # for testing - if run_once: - return Stat.get_all(r=r) - print(term.move(i + 2, 0) + term.center('[Press q to quit]')) - val = term.inkey(timeout=1) diff --git a/django_q/models.py b/django_q/models.py index dfe03d0..09b1f38 100644 --- a/django_q/models.py +++ b/django_q/models.py @@ -67,6 +67,7 @@ class Success(Task): class Meta: app_label = 'django_q' verbose_name = _('Successful task') + verbose_name_plural = _('Successful tasks') proxy = True @@ -82,6 +83,7 @@ class Failure(Task): class Meta: app_label = 'django_q' verbose_name = _('Failed task') + verbose_name_plural = _('Failed tasks') proxy = True @@ -135,4 +137,5 @@ class Schedule(models.Model): class Meta: app_label = 'django_q' verbose_name = _('Scheduled task') + verbose_name_plural = _('Scheduled tasks') ordering = ['next_run'] diff --git a/django_q/monitor.py b/django_q/monitor.py new file mode 100644 index 0000000..3e678b9 --- /dev/null +++ b/django_q/monitor.py @@ -0,0 +1,184 @@ +import socket + +# external +from blessed import Terminal + +# django +from django.core import signing +from django.utils import timezone +from django.utils.translation import ugettext as _ + +# local +from .conf import Conf, redis_client +from .tasks import SignedPackage + + +def monitor(run_once=False): + term = Terminal() + r = redis_client + with term.fullscreen(), term.hidden_cursor(), term.cbreak(): + val = None + start_width = int(term.width / 8) + while val not in (u'q', u'Q',): + col_width = int(term.width / 8) + # In case of resize + if col_width != start_width: + print(term.clear) + start_width = col_width + print(term.move(0, 0) + term.black_on_green(term.center(_('Host'), width=col_width - 1))) + print(term.move(0, 1 * col_width) + term.black_on_green(term.center(_('Id'), width=col_width - 1))) + print(term.move(0, 2 * col_width) + term.black_on_green(term.center(_('State'), width=col_width - 1))) + print(term.move(0, 3 * col_width) + term.black_on_green(term.center(_('Pool'), width=col_width - 1))) + print(term.move(0, 4 * col_width) + term.black_on_green(term.center(_('TQ'), width=col_width - 1))) + print(term.move(0, 5 * col_width) + term.black_on_green(term.center(_('RQ'), width=col_width - 1))) + print(term.move(0, 6 * col_width) + term.black_on_green(term.center(_('RC'), width=col_width - 1))) + print(term.move(0, 7 * col_width) + term.black_on_green(term.center(_('Up'), width=col_width - 1))) + i = 2 + stats = Stat.get_all(r=r) + print(term.clear_eos()) + for stat in stats: + # color status + if stat.status == Conf.WORKING: + status = term.green(Conf.WORKING) + elif stat.status == Conf.STOPPED: + status = term.red(Conf.STOPPED) + elif stat.status == Conf.IDLE: + status = Conf.IDLE + else: + status = term.yellow(stat.status) + # color q's + tasks = stat.task_q_size + if tasks > 0: + tasks = term.cyan(str(tasks)) + results = stat.done_q_size + if results > 0: + results = term.cyan(str(results)) + # color workers + workers = len(stat.workers) + if workers < Conf.WORKERS: + workers = term.yellow(str(workers)) + # format uptime + uptime = (timezone.now() - stat.tob).total_seconds() + hours, remainder = divmod(uptime, 3600) + minutes, seconds = divmod(remainder, 60) + uptime = '%d:%02d:%02d' % (hours, minutes, seconds) + # print to the terminal + print(term.move(i, 0) + term.center(stat.host[:col_width - 1], width=col_width - 1)) + print(term.move(i, 1 * col_width) + term.center(stat.cluster_id, width=col_width - 1)) + print(term.move(i, 2 * col_width) + term.center(status, width=col_width - 1)) + print(term.move(i, 3 * col_width) + term.center(workers, width=col_width - 1)) + print(term.move(i, 4 * col_width) + term.center(tasks, width=col_width - 1)) + print(term.move(i, 5 * col_width) + term.center(results, width=col_width - 1)) + print(term.move(i, 6 * col_width) + term.center(stat.reincarnations, width=col_width - 1)) + print(term.move(i, 7 * col_width) + term.center(uptime, width=col_width - 1)) + i += 1 + # for testing + if run_once: + return Stat.get_all(r=r) + print(term.move(i + 2, 0) + term.center(_('[Press q to quit]'))) + val = term.inkey(timeout=1) + + +class Status(object): + """ + Cluster status base class + """ + + def __init__(self, pid): + self.workers = [] + self.tob = None + self.reincarnations = 0 + self.cluster_id = pid + self.sentinel = 0 + self.status = 'Idle' + self.done_q_size = 0 + self.host = socket.gethostname() + self.monitor = 0 + self.task_q_size = 0 + self.pusher = 0 + self.timestamp = timezone.now() + + +class Stat(Status): + """ + Status object for Cluster monitoring + """ + + def __init__(self, sentinel): + super(Stat, self).__init__(sentinel.parent_pid) + self.r = sentinel.r + self.tob = sentinel.tob + self.reincarnations = sentinel.reincarnations + self.sentinel = sentinel.pid + self.status = sentinel.status() + self.done_q_size = sentinel.done_queue.qsize() + if sentinel.monitor: + self.monitor = sentinel.monitor.pid + self.task_q_size = sentinel.task_queue.qsize() + if sentinel.pusher: + self.pusher = sentinel.pusher.pid + for w in sentinel.pool: + self.workers.append(w.pid) + + def uptime(self): + return (timezone.now() - self.tob).total_seconds() + + @property + def key(self): + """ + :return: redis key for this cluster statistic + """ + return self.get_key(self.cluster_id) + + @staticmethod + def get_key(cluster_id): + """ + :param cluster_id: cluster ID + :return: redis key for the cluster statistic + """ + return '{}:{}'.format(Conf.Q_STAT, cluster_id) + + def save(self): + self.r.set(self.key, SignedPackage.dumps(self, True), 3) + + def empty_queues(self): + return self.done_q_size + self.task_q_size == 0 + + @staticmethod + def get(cluster_id, r=redis_client): + """ + gets the current status for the cluster + :param cluster_id: id of the cluster + :return: Stat or Status + """ + key = Stat.get_key(cluster_id) + if r.exists(key): + pack = r.get(key) + try: + return SignedPackage.loads(pack) + except signing.BadSignature: + return None + return Status(cluster_id) + + @staticmethod + def get_all(r=redis_client): + """ + Gets status for all currently running clusters with the same prefix and secret key + :return: list of type Stat + """ + stats = [] + keys = r.keys(pattern='{}:*'.format(Conf.Q_STAT)) + if keys: + packs = r.mget(keys) + for pack in packs: + try: + stats.append(SignedPackage.loads(pack)) + except signing.BadSignature: + continue + return stats + + def __getstate__(self): + # Don't pickle the redis connection + state = dict(self.__dict__) + del state['r'] + return state diff --git a/django_q/tasks.py b/django_q/tasks.py new file mode 100644 index 0000000..17f7235 --- /dev/null +++ b/django_q/tasks.py @@ -0,0 +1,119 @@ +try: + import cPickle as pickle +except ImportError: + import pickle + +# django +from django.core import signing +from django.utils import timezone + +# local +from .conf import Conf, redis_client, logger +from .models import Schedule, Task +from .humanhash import uuid + + + + +def async(func, *args, **kwargs): + """ + Sends a task to the cluster + """ + + hook = kwargs.pop('hook', None) + list_key = kwargs.pop('list_key', Conf.Q_LIST) + r = kwargs.pop('redis', redis_client) + + task = {'name': uuid()[0], 'func': func, 'hook': hook, 'args': args, 'kwargs': kwargs, 'started': timezone.now()} + pack = SignedPackage.dumps(task) + r.rpush(list_key, pack) + logger.debug('Pushed {}'.format(pack)) + return task['name'] + + +def schedule(func, *args, **kwargs): + """ + :param func: function to schedule + :param args: function arguments + :param hook: optional result hook function + :type schedule_type: Schedule.TYPE + :param repeats: how many times to repeat. 0=never, -1=always + :param next_run: Next scheduled run + :type next_run: datetime.datetime + :param kwargs: function keyword arguments + :return: the schedule object + :rtype: Schedule + """ + + hook = kwargs.pop('hook', None) + schedule_type = kwargs.pop('schedule_type', Schedule.ONCE) + repeats = kwargs.pop('repeats', -1) + next_run = kwargs.pop('next_run', timezone.now()) + + return Schedule.objects.create(func=func, + hook=hook, + args=args, + kwargs=kwargs, + schedule_type=schedule_type, + repeats=repeats, + next_run=next_run + ) + + +def result(name): + """ + Returns the result of the named task + :type name: str or unicode + :param name: the task name + :return: the result object of this task + :rtype: object or str + """ + return Task.get_result(name) + + +def fetch(name): + """ + Returns the processed task + :param name: the task name + :type name: str or unicode + :return: the full task object + :rtype: Task + """ + if Task.objects.filter(name=name).exists(): + return Task.objects.get(name=name) + + +class SignedPackage(object): + """ + Wraps Django's signing module with custom Pickle serializer + """ + + @staticmethod + def dumps(obj, compressed=Conf.COMPRESSED): + return signing.dumps(obj, + key=Conf.SECRET_KEY, + salt='django_q.q', + compress=compressed, + serializer=PickleSerializer) + + @staticmethod + def loads(obj): + return signing.loads(obj, + key=Conf.SECRET_KEY, + salt='django_q.q', + serializer=PickleSerializer) + + +class PickleSerializer(object): + """ + Simple wrapper around Pickle for signing.dumps and + signing.loads. + """ + + @staticmethod + def dumps(obj): + return pickle.dumps(obj) + + @staticmethod + def loads(data): + return pickle.loads(data) \ No newline at end of file diff --git a/django_q/tests/test_admin.py b/django_q/tests/test_admin.py index 3c59acd..9e2b525 100644 --- a/django_q/tests/test_admin.py +++ b/django_q/tests/test_admin.py @@ -4,8 +4,8 @@ from django.utils import timezone import pytest -from django_q import Task, schedule -from django_q.models import Failure +from django_q.tasks import schedule +from django_q.models import Task, Failure @pytest.mark.django_db diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index f20e948..2d62eec 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -8,11 +8,12 @@ import pytest myPath = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, myPath + '/../') -from django_q.core import Cluster, async, pusher, worker, monitor, redis_client, Sentinel +from django_q.cluster import Cluster, Sentinel, pusher, worker, monitor from django_q.humanhash import DEFAULT_WORDLIST -from django_q import result, get_task, Task -from django_q.tests.tasks import multiply -from django_q.conf import Conf +from django_q.tasks import fetch, async, result +from django_q.models import Task +from django_q.conf import Conf, redis_client +from .tasks import multiply class WordClass(object): @@ -143,40 +144,40 @@ def test_async(r): assert result_queue.qsize() == 0 # Check the results # task a - result_a = get_task(a) + result_a = fetch(a) assert result_a is not None assert result_a.success is True assert result(a) == 1506 # task b - result_b = get_task(b) + result_b = fetch(b) assert result_b is not None assert result_b.success is True assert result(b) == 1506 # task c - result_c = get_task(c) + result_c = fetch(c) assert result_c is not None assert result_c.success is False # task d - result_d = get_task(d) + result_d = fetch(d) assert result_d is not None assert result_d.success is False # task e - result_e = get_task(e) + result_e = fetch(e) assert result_e is not None assert result_e.success is True assert result(e) is None # task f - result_f = get_task(f) + result_f = fetch(f) assert result_f is not None assert result_f.success is True assert result(f) == 1506 # task g - result_g = get_task(g) + result_g = fetch(g) assert result_g is not None assert result_g.success is True assert result(g) == 'John' # task h - result_h = get_task(h) + result_h = fetch(h) assert result_h is not None assert result_h.success is True assert result(h) == 12 diff --git a/django_q/tests/test_monitor.py b/django_q/tests/test_monitor.py index d07e418..7c140b6 100644 --- a/django_q/tests/test_monitor.py +++ b/django_q/tests/test_monitor.py @@ -1,5 +1,5 @@ -from django_q.core import Cluster -from django_q.management.commands.qmonitor import monitor +from django_q.cluster import Cluster +from django_q.monitor import monitor def test_monitor(): diff --git a/django_q/tests/test_scheduler.py b/django_q/tests/test_scheduler.py index 23298fd..e10db1c 100644 --- a/django_q/tests/test_scheduler.py +++ b/django_q/tests/test_scheduler.py @@ -2,8 +2,9 @@ from multiprocessing import Queue, Event, Value import pytest -from django_q.core import scheduler, pusher, worker, monitor, redis_client, schedule as create_schedule -from django_q import Schedule, get_task +from django_q.conf import redis_client +from django_q.cluster import pusher, worker, monitor, scheduler +from django_q.tasks import Schedule, fetch, schedule as create_schedule @pytest.fixture @@ -44,7 +45,7 @@ def test_scheduler(r): assert schedule.repeats == 0 assert schedule.last_run() is not None assert schedule.success() is True - task = get_task(schedule.task) + task = fetch(schedule.task) assert task is not None assert task.success is True assert task.result < 0