From d22c916bfccc828093d9c0d60decd68f063bc3a3 Mon Sep 17 00:00:00 2001 From: Jason McVetta Date: Wed, 12 Feb 2020 17:53:52 +0700 Subject: [PATCH] add cluster_id to Sentinel --- django_q/cluster.py | 5 +++-- django_q/tests/test_cluster.py | 16 +++++++++++----- 2 files changed, 14 insertions(+), 7 deletions(-) diff --git a/django_q/cluster.py b/django_q/cluster.py index 30cb626..14766d2 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -50,7 +50,7 @@ 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)) while not self.start_event.is_set(): @@ -97,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/tests/test_cluster.py b/django_q/tests/test_cluster.py index 5e289d7..2c2fbdb 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,7 +338,8 @@ 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")