mirror of
https://github.com/django-q2/django-q2.git
synced 2026-10-05 07:18:13 +08:00
Adds django cache as monitor cache
Not the best solution yet, but it solves monitoring problems with brokers that don't have a pattern getter for cluster stats.
This commit is contained in:
@@ -1,10 +1,12 @@
|
|||||||
from django_q.conf import Conf
|
from django_q.conf import Conf
|
||||||
|
from django.core.cache import caches, InvalidCacheBackendError
|
||||||
|
|
||||||
|
|
||||||
class Broker(object):
|
class Broker(object):
|
||||||
def __init__(self, list_key=Conf.Q_LIST):
|
def __init__(self, list_key=Conf.Q_LIST):
|
||||||
self.connection = self.get_connection()
|
self.connection = self.get_connection()
|
||||||
self.list_key = list_key
|
self.list_key = list_key
|
||||||
|
self.cache=self.get_cache()
|
||||||
|
|
||||||
def enqueue(self, task):
|
def enqueue(self, task):
|
||||||
pass
|
pass
|
||||||
@@ -24,14 +26,36 @@ class Broker(object):
|
|||||||
def ping(self):
|
def ping(self):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
def set(self, key, value, timeout):
|
def set_stat(self, key, value, timeout):
|
||||||
pass
|
key_list=self.cache.get(Conf.Q_STAT, [])
|
||||||
|
if key not in key_list:
|
||||||
|
key_list.append(key)
|
||||||
|
self.cache.set(Conf.Q_STAT, key_list)
|
||||||
|
return self.cache.set(key, value, timeout)
|
||||||
|
|
||||||
def get(self, key):
|
def get_stat(self, key):
|
||||||
pass
|
return self.cache.get(key)
|
||||||
|
|
||||||
def get_pattern(self, pattern):
|
def get_stats(self, pattern):
|
||||||
pass
|
key_list = self.cache.get(Conf.Q_STAT)
|
||||||
|
if not key_list or len(key_list) == 0:
|
||||||
|
return []
|
||||||
|
stats = []
|
||||||
|
for key in key_list:
|
||||||
|
stat = self.cache.get(key)
|
||||||
|
if stat:
|
||||||
|
stats.append(stat)
|
||||||
|
else:
|
||||||
|
key_list.remove(key)
|
||||||
|
self.cache.set(Conf.Q_STAT,key_list)
|
||||||
|
return stats
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def get_cache():
|
||||||
|
try:
|
||||||
|
return caches[Conf.CACHE]
|
||||||
|
except InvalidCacheBackendError:
|
||||||
|
return None
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def get_connection():
|
def get_connection():
|
||||||
|
|||||||
@@ -26,14 +26,14 @@ class Redis(Broker):
|
|||||||
logger.error('Can not connect to Redis server.')
|
logger.error('Can not connect to Redis server.')
|
||||||
raise e
|
raise e
|
||||||
|
|
||||||
def set(self, key, value, timeout):
|
def set_stat(self, key, value, timeout):
|
||||||
self.connection.set(key, value, timeout)
|
self.connection.set(key, value, timeout)
|
||||||
|
|
||||||
def get(self, key):
|
def get_stat(self, key):
|
||||||
if self.connection.exists(key):
|
if self.connection.exists(key):
|
||||||
return self.connection.get(key)
|
return self.connection.get(key)
|
||||||
|
|
||||||
def get_pattern(self, pattern):
|
def get_stats(self, pattern):
|
||||||
keys = self.connection.keys(pattern=pattern)
|
keys = self.connection.keys(pattern=pattern)
|
||||||
if keys:
|
if keys:
|
||||||
return self.connection.mget(keys)
|
return self.connection.mget(keys)
|
||||||
|
|||||||
@@ -85,6 +85,9 @@ class Conf(object):
|
|||||||
# Global sync option to for debugging
|
# Global sync option to for debugging
|
||||||
SYNC = conf.get('sync', False)
|
SYNC = conf.get('sync', False)
|
||||||
|
|
||||||
|
# The Django cache to use
|
||||||
|
CACHE = conf.get('cache', 'default')
|
||||||
|
|
||||||
# If set to False the scheduler won't execute tasks in the past.
|
# If set to False the scheduler won't execute tasks in the past.
|
||||||
# Instead it will run once and reschedule the next run in the future. Defaults to True.
|
# Instead it will run once and reschedule the next run in the future. Defaults to True.
|
||||||
CATCH_UP = conf.get('catch_up', True)
|
CATCH_UP = conf.get('catch_up', True)
|
||||||
|
|||||||
+3
-3
@@ -64,7 +64,7 @@ class Stat(Status):
|
|||||||
|
|
||||||
def save(self):
|
def save(self):
|
||||||
try:
|
try:
|
||||||
self.broker.set(self.key, signing.SignedPackage.dumps(self, True), 3)
|
self.broker.set_stat(self.key, signing.SignedPackage.dumps(self, True), 3)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(e)
|
logger.error(e)
|
||||||
|
|
||||||
@@ -78,7 +78,7 @@ class Stat(Status):
|
|||||||
:param cluster_id: id of the cluster
|
:param cluster_id: id of the cluster
|
||||||
:return: Stat or Status
|
:return: Stat or Status
|
||||||
"""
|
"""
|
||||||
pack = broker.get(Stat.get_key(cluster_id))
|
pack = broker.get_stat(Stat.get_key(cluster_id))
|
||||||
if pack:
|
if pack:
|
||||||
try:
|
try:
|
||||||
return signing.SignedPackage.loads(pack)
|
return signing.SignedPackage.loads(pack)
|
||||||
@@ -94,7 +94,7 @@ class Stat(Status):
|
|||||||
:return: list of type Stat
|
:return: list of type Stat
|
||||||
"""
|
"""
|
||||||
stats = []
|
stats = []
|
||||||
packs = broker.get_pattern('{}:*'.format(Conf.Q_STAT)) or []
|
packs = broker.get_stats('{}:*'.format(Conf.Q_STAT)) or []
|
||||||
for pack in packs:
|
for pack in packs:
|
||||||
try:
|
try:
|
||||||
stats.append(signing.SignedPackage.loads(pack))
|
stats.append(signing.SignedPackage.loads(pack))
|
||||||
|
|||||||
Reference in New Issue
Block a user