diff --git a/django_q/__init__.py b/django_q/__init__.py index 72b57bb..c0913fb 100644 --- a/django_q/__init__.py +++ b/django_q/__init__.py @@ -7,7 +7,7 @@ sys.path.insert(0, myPath) from .tasks import async, schedule, result, result_group, fetch, fetch_group, count_group, delete_group, queue_size from .models import Task, Schedule, Success, Failure from .cluster import Cluster -from .monitor import Stat +from .status import Stat VERSION = (0, 5, 3) diff --git a/django_q/cluster.py b/django_q/cluster.py index aa41397..1e03148 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -11,7 +11,6 @@ standard_library.install_aliases() # Standard import importlib -import os import signal import socket import sys @@ -31,9 +30,9 @@ from django import db import signing import tasks -from django_q.conf import Conf, redis_client, logger, psutil +from django_q.conf import Conf, redis_client, logger, psutil, get_ppid from django_q.models import Task, Success, Schedule -from django_q.monitor import Status, Stat, ping_redis +from django_q.status import Stat, Status, ping_redis class Cluster(object): @@ -111,7 +110,7 @@ class Sentinel(object): signal.signal(signal.SIGINT, signal.SIG_IGN) signal.signal(signal.SIGTERM, signal.SIG_DFL) self.pid = current_process().pid - self.parent_pid = os.getppid() + self.parent_pid = get_ppid() self.name = current_process().name self.list_key = list_key self.r = redis_client @@ -166,7 +165,7 @@ class Sentinel(object): return p def spawn_pusher(self): - return self.spawn_process(pusher, self.task_queue, self.event_out, self.list_key, self.r) + return self.spawn_process(pusher, self.task_queue, self.event_out, self.list_key) def spawn_worker(self): self.spawn_process(worker, self.task_queue, self.result_queue, Value('f', -1), self.timeout) @@ -288,7 +287,7 @@ class Sentinel(object): Stat(self).save() -def pusher(task_queue, event, list_key=Conf.Q_LIST, r=redis_client): +def pusher(task_queue, event, list_key=Conf.Q_LIST): """ Pulls tasks of the Redis List and puts them in the task queue :type task_queue: multiprocessing.Queue @@ -296,6 +295,7 @@ def pusher(task_queue, event, list_key=Conf.Q_LIST, r=redis_client): :type list_key: str """ logger.info(_('{} pushing tasks at {}').format(current_process().name, current_process().pid)) + r = redis_client while True: try: task = r.blpop(list_key, 1) diff --git a/django_q/conf.py b/django_q/conf.py index 9fb5f6d..16a00b1 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -7,6 +7,7 @@ from django.utils.translation import ugettext_lazy as _ from django.conf import settings # external +import os import redis # optional @@ -143,3 +144,13 @@ def get_redis_client(): # redis client redis_client = get_redis_client() + + +# get parent pid compatibility +def get_ppid(): + if hasattr(os, 'getppid'): + return os.getppid() + elif psutil: + return psutil.Process(os.getpid()).ppid() + else: + raise OSError('Your OS does not support `os.getppid`. Please install `psutil` as an alternative provider.') diff --git a/django_q/monitor.py b/django_q/monitor.py index 80fbb34..9acc8b0 100644 --- a/django_q/monitor.py +++ b/django_q/monitor.py @@ -1,5 +1,4 @@ from datetime import timedelta -import socket # external from blessed import Terminal @@ -11,8 +10,8 @@ from django.utils import timezone from django.utils.translation import ugettext as _ # local -import signing -from django_q.conf import Conf, redis_client, logger +from django_q.conf import Conf, redis_client +from django_q.status import Stat, ping_redis from django_q import models @@ -85,113 +84,6 @@ def monitor(run_once=False, r=redis_client): 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 = Conf.STOPPED - 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 or sentinel.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 = 0 - self.task_q_size = 0 - if Conf.QSIZE: - self.done_q_size = sentinel.result_queue.qsize() - self.task_q_size = sentinel.task_queue.qsize() - if sentinel.monitor: - self.monitor = sentinel.monitor.pid - if sentinel.pusher: - self.pusher = sentinel.pusher.pid - self.workers = [w.pid for w in sentinel.pool] - - 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): - try: - self.r.set(self.key, signing.SignedPackage.dumps(self, True), 3) - except Exception as e: - logger.error(e) - - 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 signing.SignedPackage.loads(pack) - except signing.BadSignature: - return None - return Status(cluster_id) - - @staticmethod - def get_all(r=redis_client): - """ - Get the 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(signing.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 info(r=redis_client): term = Terminal() ping_redis(r) @@ -272,11 +164,3 @@ def info(r=redis_client): term.white('{0:.4f}'.format(exec_time)) ) return True - - -def ping_redis(r): - try: - r.ping() - except Exception as e: - logger.error('Can not connect to Redis server.') - raise e diff --git a/django_q/status.py b/django_q/status.py new file mode 100644 index 0000000..ef98148 --- /dev/null +++ b/django_q/status.py @@ -0,0 +1,119 @@ +import socket +from django.utils import timezone +from django_q.conf import Conf, logger, redis_client +import signing + + +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 = Conf.STOPPED + 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 or sentinel.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 = 0 + self.task_q_size = 0 + if Conf.QSIZE: + self.done_q_size = sentinel.result_queue.qsize() + self.task_q_size = sentinel.task_queue.qsize() + if sentinel.monitor: + self.monitor = sentinel.monitor.pid + if sentinel.pusher: + self.pusher = sentinel.pusher.pid + self.workers = [w.pid for w in sentinel.pool] + + 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): + try: + self.r.set(self.key, signing.SignedPackage.dumps(self, True), 3) + except Exception as e: + logger.error(e) + + 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 signing.SignedPackage.loads(pack) + except signing.BadSignature: + return None + return Status(cluster_id) + + @staticmethod + def get_all(r=redis_client): + """ + Get the 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(signing.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 ping_redis(r): + try: + r.ping() + except Exception as e: + logger.error('Can not connect to Redis server.') + raise e diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index a5c1154..7e5a407 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -14,7 +14,7 @@ from django_q.humanhash import DEFAULT_WORDLIST from django_q.tasks import fetch, fetch_group, async, result, result_group, count_group, delete_group, queue_size from django_q.models import Task, Success from django_q.conf import Conf, redis_client -from django_q.monitor import Stat +from django_q.status import Stat from .tasks import multiply @@ -85,7 +85,7 @@ def test_cluster(r): event = Event() event.set() # Test push - pusher(task_queue, event, list_key=list_key, r=r) + pusher(task_queue, event, list_key=list_key) assert task_queue.qsize() == 1 assert queue_size(list_key=list_key, r=r) == 0 # Test work @@ -148,7 +148,7 @@ def test_async(r, admin_user): stop_event.set() # push the tasks for i in range(task_count): - pusher(task_queue, stop_event, list_key=list_key, r=r) + pusher(task_queue, stop_event, list_key=list_key) assert queue_size(list_key=list_key, r=r) == 0 assert task_queue.qsize() == task_count task_queue.put('STOP') @@ -292,8 +292,8 @@ def test_recycle(r): task_queue = Queue() result_queue = Queue() # push two tasks - pusher(task_queue, stop_event, list_key=list_key, r=r) - pusher(task_queue, stop_event, list_key=list_key, r=r) + pusher(task_queue, stop_event, list_key=list_key) + pusher(task_queue, stop_event, list_key=list_key) # worker should exit on recycle worker(task_queue, result_queue, Value('f', -1)) # check if the work has been done @@ -322,7 +322,7 @@ def test_bad_secret(r, monkeypatch): assert len(stat) == 0 assert Stat.get(s.parent_pid, r) is None task_queue = Queue() - pusher(task_queue, stop_event, list_key=list_key, r=r) + pusher(task_queue, stop_event, list_key=list_key) result_queue = Queue() task_queue.put('STOP') worker(task_queue, result_queue, Value('f', -1), ) diff --git a/django_q/tests/test_monitor.py b/django_q/tests/test_monitor.py index 9a266b0..4f94b75 100644 --- a/django_q/tests/test_monitor.py +++ b/django_q/tests/test_monitor.py @@ -3,7 +3,8 @@ import redis from django_q import async from django_q.cluster import Cluster -from django_q.monitor import monitor, Stat, ping_redis, info +from django_q.monitor import monitor, info +from django_q.status import Stat, ping_redis @pytest.mark.django_db diff --git a/django_q/tests/test_scheduler.py b/django_q/tests/test_scheduler.py index 05f5281..7709477 100644 --- a/django_q/tests/test_scheduler.py +++ b/django_q/tests/test_scheduler.py @@ -34,7 +34,7 @@ def test_scheduler(r): stop_event = Event() stop_event.set() # push it - pusher(task_queue, stop_event, list_key=list_key, r=r) + pusher(task_queue, stop_event, list_key=list_key) assert task_queue.qsize() == 1 assert queue_size(list_key=list_key, r=r) == 0 task_queue.put('STOP')