From 439c8a2f7728c0a282d4afbd2e6a4a15cc555214 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Wed, 26 Aug 2015 13:11:58 +0200 Subject: [PATCH] Moves Status to it's own file This way only qmonitor and qinfo need Curses and you can still run the cluster without it. --- django_q/__init__.py | 2 +- django_q/cluster.py | 2 +- django_q/monitor.py | 120 +-------------------------------- django_q/status.py | 119 ++++++++++++++++++++++++++++++++ django_q/tests/test_cluster.py | 2 +- django_q/tests/test_monitor.py | 3 +- 6 files changed, 126 insertions(+), 122 deletions(-) create mode 100644 django_q/status.py 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 cd01887..1e03148 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -32,7 +32,7 @@ import tasks 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): 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 8ca9e3a..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 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