mirror of
https://github.com/django-q2/django-q2.git
synced 2026-10-07 12:08:11 +08:00
use both pid and cluster_id
This commit is contained in:
+2
-2
@@ -76,8 +76,8 @@ class Cluster(object):
|
|||||||
@property
|
@property
|
||||||
def stat(self):
|
def stat(self):
|
||||||
if self.sentinel:
|
if self.sentinel:
|
||||||
return Stat.get(self.pid)
|
return Stat.get(pid=self.pid, cluster_id=self.cluster_id)
|
||||||
return Status(self.pid)
|
return Status(pid=self.pid, cluster_id=self.cluster_id)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def is_starting(self):
|
def is_starting(self):
|
||||||
|
|||||||
+6
-5
@@ -8,11 +8,12 @@ from django_q.signing import SignedPackage, BadSignature
|
|||||||
class Status(object):
|
class Status(object):
|
||||||
"""Cluster status base class."""
|
"""Cluster status base class."""
|
||||||
|
|
||||||
def __init__(self, pid):
|
def __init__(self, pid, cluster_id):
|
||||||
self.workers = []
|
self.workers = []
|
||||||
self.tob = None
|
self.tob = None
|
||||||
self.reincarnations = 0
|
self.reincarnations = 0
|
||||||
self.cluster_id = pid
|
self.pid = pid
|
||||||
|
self.cluster_id = cluster_id
|
||||||
self.sentinel = 0
|
self.sentinel = 0
|
||||||
self.status = Conf.STOPPED
|
self.status = Conf.STOPPED
|
||||||
self.done_q_size = 0
|
self.done_q_size = 0
|
||||||
@@ -27,7 +28,7 @@ class Stat(Status):
|
|||||||
"""Status object for Cluster monitoring."""
|
"""Status object for Cluster monitoring."""
|
||||||
|
|
||||||
def __init__(self, sentinel):
|
def __init__(self, sentinel):
|
||||||
super(Stat, self).__init__(sentinel.parent_pid or sentinel.pid)
|
super(Stat, self).__init__(sentinel.parent_pid or sentinel.pid, cluster_id=sentinel.cluster_id)
|
||||||
self.broker = sentinel.broker or get_broker()
|
self.broker = sentinel.broker or get_broker()
|
||||||
self.tob = sentinel.tob
|
self.tob = sentinel.tob
|
||||||
self.reincarnations = sentinel.reincarnations
|
self.reincarnations = sentinel.reincarnations
|
||||||
@@ -72,7 +73,7 @@ class Stat(Status):
|
|||||||
return self.done_q_size + self.task_q_size == 0
|
return self.done_q_size + self.task_q_size == 0
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def get(cluster_id, broker=None):
|
def get(pid, cluster_id, broker=None):
|
||||||
"""
|
"""
|
||||||
gets the current status for the cluster
|
gets the current status for the cluster
|
||||||
:param cluster_id: id of the cluster
|
:param cluster_id: id of the cluster
|
||||||
@@ -86,7 +87,7 @@ class Stat(Status):
|
|||||||
return SignedPackage.loads(pack)
|
return SignedPackage.loads(pack)
|
||||||
except BadSignature:
|
except BadSignature:
|
||||||
return None
|
return None
|
||||||
return Status(cluster_id)
|
return Status(pid=pid, cluster_id=cluster_id)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def get_all(broker=None):
|
def get_all(broker=None):
|
||||||
|
|||||||
@@ -345,7 +345,7 @@ def test_bad_secret(broker, monkeypatch):
|
|||||||
monkeypatch.setattr(Conf, "SECRET_KEY", "OOPS")
|
monkeypatch.setattr(Conf, "SECRET_KEY", "OOPS")
|
||||||
stat = Stat.get_all()
|
stat = Stat.get_all()
|
||||||
assert len(stat) == 0
|
assert len(stat) == 0
|
||||||
assert Stat.get(s.parent_pid) is None
|
assert Stat.get(pid=s.parent_pid, cluster_id=cluster_id) is None
|
||||||
task_queue = Queue()
|
task_queue = Queue()
|
||||||
pusher(task_queue, stop_event, broker=broker)
|
pusher(task_queue, stop_event, broker=broker)
|
||||||
result_queue = Queue()
|
result_queue = Queue()
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
import pytest
|
import pytest
|
||||||
|
import uuid
|
||||||
|
|
||||||
from django_q.tasks import async_task
|
from django_q.tasks import async_task
|
||||||
from django_q.brokers import get_broker
|
from django_q.brokers import get_broker
|
||||||
@@ -10,7 +11,8 @@ from django_q.conf import Conf
|
|||||||
|
|
||||||
@pytest.mark.django_db
|
@pytest.mark.django_db
|
||||||
def test_monitor(monkeypatch):
|
def test_monitor(monkeypatch):
|
||||||
assert Stat.get(0).sentinel == 0
|
cluster_id = uuid.uuid4()
|
||||||
|
assert Stat.get(pid=0, cluster_id=4).sentinel == 0
|
||||||
c = Cluster()
|
c = Cluster()
|
||||||
c.start()
|
c.start()
|
||||||
stats = monitor(run_once=True)
|
stats = monitor(run_once=True)
|
||||||
@@ -18,7 +20,7 @@ def test_monitor(monkeypatch):
|
|||||||
assert len(stats) > 0
|
assert len(stats) > 0
|
||||||
found_c = False
|
found_c = False
|
||||||
for stat in stats:
|
for stat in stats:
|
||||||
if stat.cluster_id == c.pid:
|
if stat.cluster_id == c.cluster_id:
|
||||||
found_c = True
|
found_c = True
|
||||||
assert stat.uptime() > 0
|
assert stat.uptime() > 0
|
||||||
assert stat.empty_queues() is True
|
assert stat.empty_queues() is True
|
||||||
|
|||||||
Reference in New Issue
Block a user