diff --git a/django_q/cluster.py b/django_q/cluster.py index 860527c..bece2df 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -14,6 +14,7 @@ import importlib import signal import socket import traceback +import uuid # Django from django import db from django.utils import timezone @@ -38,6 +39,7 @@ class Cluster(object): self.stop_event = None self.start_event = None self.pid = current_process().pid + self.cluster_id = uuid.uuid4() self.host = socket.gethostname() self.timeout = Conf.TIMEOUT signal.signal(signal.SIGTERM, self.sig_handler) @@ -48,9 +50,9 @@ class Cluster(object): self.stop_event = Event() self.start_event = Event() self.sentinel = Process(target=Sentinel, - args=(self.stop_event, self.start_event, self.broker, self.timeout)) + args=(self.stop_event, self.start_event, self.cluster_id, self.broker, self.timeout)) self.sentinel.start() - logger.info(_('Q Cluster-{} starting.').format(self.pid)) + logger.info(_('Q Cluster-{} starting.').format(self.cluster_id)) while not self.start_event.is_set(): sleep(0.1) return self.pid @@ -58,10 +60,10 @@ class Cluster(object): def stop(self): if not self.sentinel.is_alive(): return False - logger.info(_('Q Cluster-{} stopping.').format(self.pid)) + logger.info(_('Q Cluster-{} stopping.').format(self.cluster_id)) self.stop_event.set() self.sentinel.join() - logger.info(_('Q Cluster-{} has stopped.').format(self.pid)) + logger.info(_('Q Cluster-{} has stopped.').format(self.cluster_id)) self.start_event = None self.stop_event = None return True @@ -74,8 +76,8 @@ class Cluster(object): @property def stat(self): if self.sentinel: - return Stat.get(self.pid) - return Status(self.pid) + return Stat.get(pid=self.pid, cluster_id=self.cluster_id) + return Status(pid=self.pid, cluster_id=self.cluster_id) @property def is_starting(self): @@ -95,11 +97,12 @@ class Cluster(object): class Sentinel(object): - def __init__(self, stop_event, start_event, broker=None, timeout=Conf.TIMEOUT, start=True): + def __init__(self, stop_event, start_event, cluster_id, broker=None, timeout=Conf.TIMEOUT, start=True): # Make sure we catch signals for the pool signal.signal(signal.SIGINT, signal.SIG_IGN) signal.signal(signal.SIGTERM, signal.SIG_DFL) self.pid = current_process().pid + self.cluster_id = cluster_id self.parent_pid = get_ppid() self.name = current_process().name self.broker = broker or get_broker() diff --git a/django_q/monitor.py b/django_q/monitor.py index 8fd233d..bccb3ac 100644 --- a/django_q/monitor.py +++ b/django_q/monitor.py @@ -72,7 +72,7 @@ def monitor(run_once=False, broker=None): uptime = '%d:%02d:%02d' % (hours, minutes, seconds) # print to the terminal print(term.move(i, 0) + term.center(stat.host[:col_width - 1], width=col_width - 1)) - print(term.move(i, 1 * col_width) + term.center(stat.cluster_id, width=col_width - 1)) + print(term.move(i, 1 * col_width) + term.center(str(stat.cluster_id)[-8:], width=col_width - 1)) print(term.move(i, 2 * col_width) + term.center(status, width=col_width - 1)) print(term.move(i, 3 * col_width) + term.center(workers, width=col_width - 1)) print(term.move(i, 4 * col_width) + term.center(tasks, width=col_width - 1)) diff --git a/django_q/status.py b/django_q/status.py index 51f32ee..06402a9 100644 --- a/django_q/status.py +++ b/django_q/status.py @@ -8,11 +8,12 @@ from django_q.signing import SignedPackage, BadSignature class Status(object): """Cluster status base class.""" - def __init__(self, pid): + def __init__(self, pid, cluster_id): self.workers = [] self.tob = None self.reincarnations = 0 - self.cluster_id = pid + self.pid = pid + self.cluster_id = cluster_id self.sentinel = 0 self.status = Conf.STOPPED self.done_q_size = 0 @@ -27,7 +28,7 @@ class Stat(Status): """Status object for Cluster monitoring.""" 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.tob = sentinel.tob self.reincarnations = sentinel.reincarnations @@ -72,7 +73,7 @@ class Stat(Status): return self.done_q_size + self.task_q_size == 0 @staticmethod - def get(cluster_id, broker=None): + def get(pid, cluster_id, broker=None): """ gets the current status for the cluster :param cluster_id: id of the cluster @@ -86,7 +87,7 @@ class Stat(Status): return SignedPackage.loads(pack) except BadSignature: return None - return Status(cluster_id) + return Status(pid=pid, cluster_id=cluster_id) @staticmethod def get_all(broker=None): diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 5e289d7..f3905f9 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -3,6 +3,7 @@ import threading from multiprocessing import Event, Value from time import sleep from django.utils import timezone +import uuid as uuidlib import os import pytest @@ -72,7 +73,8 @@ def test_sentinel(): start_event = Event() stop_event = Event() stop_event.set() - s = Sentinel(stop_event, start_event, broker=get_broker('sentinel_test:q')) + cluster_id = uuidlib.uuid4() + s = Sentinel(stop_event, start_event, cluster_id=cluster_id, broker=get_broker('sentinel_test:q')) assert start_event.is_set() assert s.status() == Conf.STOPPED @@ -256,9 +258,10 @@ def test_timeout(broker, cluster_config_timeout, async_task_kwargs): async_task('time.sleep', 5, broker=broker, **async_task_kwargs) start_event = Event() stop_event = Event() + cluster_id = uuidlib.uuid4() # Set a timer to stop the Sentinel threading.Timer(3, stop_event.set).start() - s = Sentinel(stop_event, start_event, broker=broker, timeout=cluster_config_timeout) + s = Sentinel(stop_event, start_event, cluster_id=cluster_id, broker=broker, timeout=cluster_config_timeout) assert start_event.is_set() assert s.status() == Conf.STOPPED assert s.reincarnations == 1 @@ -279,9 +282,10 @@ def test_timeout_task_finishes(broker, cluster_config_timeout, async_task_kwargs async_task('time.sleep', 3, broker=broker, **async_task_kwargs) start_event = Event() stop_event = Event() + cluster_id = uuidlib.uuid4() # Set a timer to stop the Sentinel threading.Timer(6, stop_event.set).start() - s = Sentinel(stop_event, start_event, broker=broker, timeout=cluster_config_timeout) + s = Sentinel(stop_event, start_event, cluster_id=cluster_id, broker=broker, timeout=cluster_config_timeout) assert start_event.is_set() assert s.status() == Conf.STOPPED assert s.reincarnations == 0 @@ -297,12 +301,13 @@ def test_recycle(broker, monkeypatch): async_task('django_q.tests.tasks.multiply', 2, 2, broker=broker) start_event = Event() stop_event = Event() + cluster_id = uuidlib.uuid4() # override settings monkeypatch.setattr(Conf, 'RECYCLE', 2) monkeypatch.setattr(Conf, 'WORKERS', 1) # set a timer to stop the Sentinel threading.Timer(3, stop_event.set).start() - s = Sentinel(stop_event, start_event, broker=broker) + s = Sentinel(stop_event, start_event, cluster_id=cluster_id, broker=broker) assert start_event.is_set() assert s.status() == Conf.STOPPED assert s.reincarnations == 1 @@ -333,13 +338,14 @@ def test_bad_secret(broker, monkeypatch): stop_event = Event() stop_event.set() start_event = Event() - s = Sentinel(stop_event, start_event, broker=broker, start=False) + cluster_id = uuidlib.uuid4() + s = Sentinel(stop_event, start_event, cluster_id=cluster_id, broker=broker, start=False) Stat(s).save() # change the SECRET monkeypatch.setattr(Conf, "SECRET_KEY", "OOPS") stat = Stat.get_all() 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() pusher(task_queue, stop_event, broker=broker) result_queue = Queue() diff --git a/django_q/tests/test_monitor.py b/django_q/tests/test_monitor.py index 5719973..5964a31 100644 --- a/django_q/tests/test_monitor.py +++ b/django_q/tests/test_monitor.py @@ -1,4 +1,5 @@ import pytest +import uuid from django_q.tasks import async_task from django_q.brokers import get_broker @@ -10,7 +11,8 @@ from django_q.conf import Conf @pytest.mark.django_db def test_monitor(monkeypatch): - assert Stat.get(0).sentinel == 0 + cluster_id = uuid.uuid4() + assert Stat.get(pid=0, cluster_id=cluster_id).sentinel == 0 c = Cluster() c.start() stats = monitor(run_once=True) @@ -18,7 +20,7 @@ def test_monitor(monkeypatch): assert len(stats) > 0 found_c = False for stat in stats: - if stat.cluster_id == c.pid: + if stat.cluster_id == c.cluster_id: found_c = True assert stat.uptime() > 0 assert stat.empty_queues() is True