mirror of
https://github.com/django-q2/django-q2.git
synced 2026-09-24 21:28:13 +08:00
Ilan Steemers
439c8a2f77
This way only qmonitor and qinfo need Curses and you can still run the cluster without it.
120 lines
3.3 KiB
Python
120 lines
3.3 KiB
Python
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
|