diff --git a/django_q/apps.py b/django_q/apps.py index 3848062..72705fd 100644 --- a/django_q/apps.py +++ b/django_q/apps.py @@ -1,7 +1,7 @@ from django.apps import AppConfig -from .conf import LABEL +from .conf import Conf class DjangoQConfig(AppConfig): name = 'django_q' - verbose_name = LABEL + verbose_name = Conf.LABEL diff --git a/django_q/conf.py b/django_q/conf.py index 35924d3..4b2dfd6 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -5,52 +5,55 @@ from django.conf import settings VERSION = '0.1.0' -try: - conf = settings.Q_CLUSTER -except AttributeError: - conf = {} -# Redis server configuration . Follows standard redis keywords -REDIS = conf.get('redis', {}) +class Conf(object): + try: + conf = settings.Q_CLUSTER + except AttributeError: + conf = {} -# Name of the cluster or site. For when you run multiple sites on one redis server -PREFIX = conf.get('name', 'default') + # Redis server configuration . Follows standard redis keywords + REDIS = conf.get('redis', {}) -# Log output level -LOG_LEVEL = conf.get('log_level', 'INFO') + # Name of the cluster or site. For when you run multiple sites on one redis server + PREFIX = conf.get('name', 'default') -# Maximum number of successful tasks kept in the database. 0 saves everything. -1 saves none -# Failures are always saved -SAVE_LIMIT = conf.get('save_limit', 250) + # Log output level + LOG_LEVEL = conf.get('log_level', 'INFO') -# Number of workers in the pool. Default is cpu count. +2 for monitor and pusher -WORKERS = conf.get('workers', cpu_count()) + # Maximum number of successful tasks kept in the database. 0 saves everything. -1 saves none + # Failures are always saved + SAVE_LIMIT = conf.get('save_limit', 250) -# Sets compression of redis packages -COMPRESSED = conf.get('compress', False) + # Number of workers in the pool. Default is cpu count. +2 for monitor and pusher + WORKERS = conf.get('workers', cpu_count()) -# Number of tasks each worker can handle before it gets recycled. Useful for releasing memory -RECYCLE = conf.get('recycle', 500) + # Sets compression of redis packages + COMPRESSED = conf.get('compress', False) -# The Django Admin label for this app -LABEL = conf.get('label', 'Django Q') + # Number of tasks each worker can handle before it gets recycled. Useful for releasing memory + RECYCLE = conf.get('recycle', 500) -# Use the secret key for package signing -try: - SECRET_KEY = settings.SECRET_KEY -except AttributeError: - SECRET_KEY = 'omgicantbelieveudonthaveasecretkey' + # The Django Admin label for this app + LABEL = conf.get('label', 'Django Q') -# Getting the signal names -SIGNAL_NAMES = dict((getattr(signal, n), n) for n in dir(signal) if n.startswith('SIG') and '_' not in n) + # Use the secret key for package signing + try: + SECRET_KEY = settings.SECRET_KEY + except AttributeError: + SECRET_KEY = 'omgicantbelieveudonthaveasecretkey' -# The redis list key -Q_LIST = 'django_q:{}:q'.format(PREFIX) -# The redis stats key -Q_STAT = 'django_q:{}:cluster'.format(PREFIX) + # The redis list key + Q_LIST = 'django_q:{}:q'.format(PREFIX) + # The redis stats key + Q_STAT = 'django_q:{}:cluster'.format(PREFIX) -# Cluster status -STARTING = 'Starting' -RUNNING = 'Running' -STOPPED = 'Stopped' -STOPPING = 'Stopping' + # 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' diff --git a/django_q/core.py b/django_q/core.py index 1550826..4ba6bf2 100644 --- a/django_q/core.py +++ b/django_q/core.py @@ -34,8 +34,7 @@ from django.core import signing from django.utils import timezone # Local -from .conf import LOG_LEVEL, SECRET_KEY, SAVE_LIMIT, WORKERS, COMPRESSED, REDIS, Q_LIST, SIGNAL_NAMES, STARTING, \ - RUNNING, STOPPING, STOPPED, RECYCLE, Q_STAT +from .conf import Conf from .humanhash import uuid from .models import Task, Success, Schedule @@ -43,18 +42,18 @@ logger = logging.getLogger('django-q') # Set up standard logging handler in case there is none if not logger.handlers: - logger.setLevel(level=getattr(logging, LOG_LEVEL)) + 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(**REDIS) +redis_client = redis.StrictRedis(**Conf.REDIS) class Cluster(object): - def __init__(self, list_key=Q_LIST): + def __init__(self, list_key=Conf.Q_LIST): try: redis_client.ping() except Exception as e: @@ -99,7 +98,7 @@ class Cluster(object): return True def sig_handler(self, signum, frame): - logger.debug('{} got signal {}'.format(current_process().name, SIGNAL_NAMES.get(signum, 'UNKNOWN'))) + logger.debug('{} got signal {}'.format(current_process().name, Conf.SIGNAL_NAMES.get(signum, 'UNKNOWN'))) self.stop() @property @@ -130,7 +129,7 @@ class Cluster(object): class Sentinel(object): - def __init__(self, stop_event, start_event, list_key=Q_LIST, start=True): + def __init__(self, stop_event, start_event, list_key=Conf.Q_LIST, start=True): # Make sure we catch signals for the pool signal.signal(signal.SIGINT, signal.SIG_IGN) signal.signal(signal.SIGTERM, signal.SIG_DFL) @@ -139,18 +138,17 @@ class Sentinel(object): self.name = current_process().name self.list_key = list_key self.r = redis_client - self.status = None self.reincarnations = 0 self.tob = timezone.now() self.stop_event = stop_event self.start_event = start_event - self.pool_size = WORKERS + self.pool_size = Conf.WORKERS self.pool = [] self.task_queue = Queue() self.done_queue = Queue() self.event_out = Event() - self.monitor_pid = None - self.pusher_pid = None + self.monitor = None + self.pusher = None if start: self.start() @@ -158,7 +156,22 @@ class Sentinel(object): self.spawn_cluster() self.guard() + def status(self): + if not self.start_event.is_set() and not self.stop_event.is_set(): + return Conf.STARTING + elif self.start_event.is_set() and not self.stop_event.is_set(): + if self.done_queue.qsize() == 0 and self.task_queue.qsize() == 0: + return Conf.IDLE + return Conf.WORKING + elif self.stop_event.is_set() and self.start_event.is_set(): + if self.monitor.is_alive() or self.pusher.is_alive() or len(self.pool) > 0: + return Conf.STOPPING + return Conf.STOPPED + def spawn_process(self, target, *args): + """ + :type target: function or class + """ # This is just for PyCharm to not crash. Ignore it. if not hasattr(sys.stdin, 'close'): def dummy_close(): @@ -167,9 +180,10 @@ class Sentinel(object): sys.stdin.close = dummy_close p = Process(target=target, args=args) p.daemon = True - self.pool.append(p) + if target == worker: + self.pool.append(p) p.start() - return p.pid + return p def spawn_pusher(self): return self.spawn_process(pusher, self.task_queue, self.event_out, self.list_key, self.r) @@ -181,11 +195,11 @@ class Sentinel(object): return self.spawn_process(monitor, self.done_queue) def reincarnate(self, pid): - if pid == self.monitor_pid: - self.spawn_monitor() + if pid == self.monitor.pid: + self.monitor = self.spawn_monitor() logger.warn("reincarnated monitor after death of {}".format(pid)) - elif pid == self.pusher_pid: - self.spawn_pusher() + elif pid == self.pusher.pid: + self.pusher = self.spawn_pusher() logger.warn("reincarnated pusher after death of {}".format(pid)) else: self.spawn_worker() @@ -193,60 +207,72 @@ class Sentinel(object): self.reincarnations += 1 def spawn_cluster(self): - self.set_status(STARTING) + Stat(self).save() for i in range(self.pool_size): self.spawn_worker() - self.monitor_pid = self.spawn_monitor() - self.pusher_pid = self.spawn_pusher() + self.monitor = self.spawn_monitor() + self.pusher = self.spawn_pusher() def guard(self): logger.info('{} guarding cluster at {}'.format(current_process().name, self.pid)) self.start_event.set() - self.set_status(RUNNING) + Stat(self).save() logger.info('Q Cluster-{} running.'.format(self.parent_pid)) + scheduler(list_key=self.list_key) counter = 0 - while True: + # Guard loop. Runs at least once + while not self.stop_event.is_set() or not counter: + # Check Workers for p in list(self.pool): if not p.is_alive(): p.terminate() self.pool.remove(p) self.reincarnate(p.pid) - Stat(self).save() - if self.stop_event.is_set(): - break + # Check Monitor + if not self.monitor.is_alive(): + self.reincarnate(self.monitor.pid) + # Check Pusher + if not self.pusher.is_alive(): + self.reincarnate(self.monitor.pid) # Call scheduler once a minute (or so) counter += 1 - if counter > 30: + if counter > 60: counter = 0 - scheduler() - sleep(2) + scheduler(list_key=self.list_key) + # Save current status + Stat(self).save() + sleep(1) self.stop() def stop(self): - self.set_status(STOPPING) + Stat(self).save() name = current_process().name logger.info('{} stopping pool processes'.format(name)) # Stopping pusher self.event_out.set() + while self.pusher.is_alive(): + sleep(0.2) + Stat(self).save() # Putting poison pills in the queue - for _ in self.pool: + for _ in range(self.pool_size): self.task_queue.put('STOP') - while len(self.pool) > 2: - for p in list(self.pool): + # Wait for all the workers to exit + while len(self.pool) > 0: + for p in self.pool: if not p.is_alive(): logger.debug('{} stopped gracefully'.format(p.pid)) self.pool.remove(p) sleep(0.2) + Stat(self).save() # Finally stop the monitor self.done_queue.put('STOP') + while self.status() != Conf.STOPPED: + sleep(0.2) + Stat(self).save() self.pool = [] - self.set_status(STOPPED) - - def set_status(self, message=None): - Stat(self, message).save() -def pusher(task_queue, e, list_key=Q_LIST, r=redis_client): +def pusher(task_queue, e, list_key=Conf.Q_LIST, r=redis_client): """ Pulls tasks of the Redis List and puts them in the task queue :type task_queue: multiprocessing.Queue @@ -325,7 +351,7 @@ def worker(task_queue, done_queue): task['stopped'] = timezone.now() done_queue.put(task) # Recycle - if task_count == RECYCLE: + if task_count == Conf.RECYCLE and task_queue.qsize() == 0: break logger.info('{} stopped doing work'.format(name)) @@ -335,10 +361,10 @@ def save_task(task): Saves the task package to Django """ # SAVE LIMIT < 0 : Don't save success - if SAVE_LIMIT < 0 and task['success']: + if Conf.SAVE_LIMIT < 0 and task['success']: return # SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning - if task['success'] and 0 < SAVE_LIMIT < Success.objects.count(): + if task['success'] and 0 < Conf.SAVE_LIMIT < Success.objects.count(): Success.objects.first().delete() try: @@ -370,7 +396,7 @@ def async(func, *args, **kwargs): list_key = kwargs['list_key'] del kwargs['list_key'] else: - list_key = Q_LIST + list_key = Conf.Q_LIST # Check for redis connection override if 'redis' in kwargs: r = kwargs['redis'] @@ -390,9 +416,9 @@ class SignedPackage(object): """ @staticmethod - def dumps(obj, compressed=COMPRESSED): + def dumps(obj, compressed=Conf.COMPRESSED): return signing.dumps(obj, - key=SECRET_KEY, + key=Conf.SECRET_KEY, salt='django_q.q', compress=compressed, serializer=PickleSerializer) @@ -400,7 +426,7 @@ class SignedPackage(object): @staticmethod def loads(obj): return signing.loads(obj, - key=SECRET_KEY, + key=Conf.SECRET_KEY, salt='django_q.q', serializer=PickleSerializer) @@ -445,19 +471,19 @@ class Stat(Status): Status object for Cluster monitoring """ - def __init__(self, sentinel, message=None): + def __init__(self, sentinel): super(Stat, self).__init__(sentinel.parent_pid) - if message: - sentinel.status = message self.r = sentinel.r self.tob = sentinel.tob self.reincarnations = sentinel.reincarnations self.sentinel = sentinel.pid - self.status = sentinel.status + self.status = sentinel.status() self.done_q_size = sentinel.done_queue.qsize() - self.monitor = sentinel.monitor_pid + if sentinel.monitor: + self.monitor = sentinel.monitor.pid self.task_q_size = sentinel.task_queue.qsize() - self.pusher = sentinel.pusher_pid + if sentinel.pusher: + self.pusher = sentinel.pusher.pid for w in sentinel.pool: self.workers.append(w.pid) @@ -477,7 +503,7 @@ class Stat(Status): :param cluster_id: cluster ID :return: redis key for the cluster statistic """ - return '{}:{}'.format(Q_STAT, cluster_id) + return '{}:{}'.format(Conf.Q_STAT, cluster_id) def save(self): self.r.set(self.key, SignedPackage.dumps(self, True), 3) @@ -508,7 +534,7 @@ class Stat(Status): :return: Stat list """ stats = [] - keys = r.keys(pattern='{}:*'.format(Q_STAT)) + keys = r.keys(pattern='{}:*'.format(Conf.Q_STAT)) if keys: packs = r.mget(keys) for pack in packs: @@ -525,7 +551,7 @@ class Stat(Status): return state -def scheduler(): +def scheduler(list_key=Conf.Q_LIST): """ Creates a task from a schedule at the scheduled time and schedules next run """ @@ -566,6 +592,7 @@ def scheduler(): else: schedule.repeats = 0 # send it to the cluster + kwargs['list_key'] = list_key schedule.task = async(schedule.func, *args, **kwargs) if not schedule.task: logger.error('{} failed to create task from schedule {}').format(current_process().name, schedule.id) diff --git a/django_q/management/commands/qmonitor.py b/django_q/management/commands/qmonitor.py index 7e1a1bf..ee145d3 100644 --- a/django_q/management/commands/qmonitor.py +++ b/django_q/management/commands/qmonitor.py @@ -6,7 +6,8 @@ from django.utils import timezone from blessed import Terminal # Local -from django_q.core import Stat, RUNNING, STOPPED, redis_client +from django_q.core import Stat, redis_client +from django_q.conf import Conf # TODO add name argument to monitor different clusters @@ -42,23 +43,37 @@ def monitor(run_once=False): print(term.clear_eos()) for stat in stats: # color status - if stat.status == RUNNING: - status = term.green(RUNNING) - elif stat.status == STOPPED: - status = term.red(STOPPED) + 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(len(stat.workers), width=col_width - 1)) - print(term.move(i, 4 * col_width) + term.center(stat.task_q_size, width=col_width - 1)) - print(term.move(i, 5 * col_width) + term.center(stat.done_q_size, 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 diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 2bfffd9..1e83756 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -4,12 +4,14 @@ from multiprocessing import Queue, Event import pytest +from conf import Conf + 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.core import Cluster, async, pusher, worker, monitor, redis_client, Sentinel, scheduler from django_q.humanhash import DEFAULT_WORDLIST -from django_q import result, get_task, Task +from django_q import result, get_task, Task, Schedule from django_q.tests.tasks import multiply @@ -30,6 +32,7 @@ def test_redis_connection(r): assert r.ping() is True +@pytest.mark.django_db def test_cluster_initial(): c = Cluster() assert c.sentinel is None @@ -37,17 +40,21 @@ def test_cluster_initial(): assert c.start() > 0 assert c.sentinel.is_alive() is True assert c.is_running + stat = c.stat + assert stat.status == Conf.IDLE assert c.stop() is True assert c.sentinel.is_alive() is False assert c.has_stopped +@pytest.mark.django_db def test_sentinel(): start_event = Event() stop_event = Event() stop_event.set() - Sentinel(stop_event, start_event, list_key='sentinel_test:q') + s = Sentinel(stop_event, start_event, list_key='sentinel_test:q') assert start_event.is_set() + assert s.status() == Conf.STOPPED @pytest.mark.django_db @@ -165,7 +172,6 @@ def test_async(r): r.delete(list_key) -# not sure if this actually asserts, but it is called @pytest.mark.django_db def assert_result(task): assert task is not None diff --git a/django_q/tests/test_scheduler.py b/django_q/tests/test_scheduler.py new file mode 100644 index 0000000..51c1464 --- /dev/null +++ b/django_q/tests/test_scheduler.py @@ -0,0 +1,46 @@ +from multiprocessing import Queue, Event +import pytest +from django_q.core import scheduler, pusher, worker, monitor, redis_client +from django_q import Schedule, get_task + +@pytest.fixture +def r(): + return redis_client + +@pytest.mark.django_db +def test_scheduler(r): + list_key = 'scheduler_test:q' + r.delete(list_key) + schedule = Schedule.objects.create(func='math.copysign', + args='1, -1', + schedule_type=Schedule.ONCE, + repeats=1, + hook='django_q.tests.tasks.result' + ) + assert schedule.last_run() is None + # run scheduler + scheduler(list_key=list_key) + # set up the workflow + task_queue = Queue() + stop_event = Event() + stop_event.set() + # push it + pusher(task_queue, stop_event, list_key=list_key, r=r) + assert task_queue.qsize() == 1 + assert r.llen(list_key) == 0 + task_queue.put('STOP') + # let a worker handle them + result_queue = Queue() + worker(task_queue, result_queue) + assert result_queue.qsize() == 1 + result_queue.put('STOP') + # store the results + monitor(result_queue) + assert result_queue.qsize() == 0 + schedule.refresh_from_db() + assert schedule.repeats == 0 + assert schedule.success() is True + task = get_task(schedule.task) + assert task is not None + assert task.success is True + assert task.result < 0