fork/django-q2
1
0
Fork 0
mirror of https://github.com/django-q2/django-q2.git synced 2026-10-02 20:08:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

add cluster_id to Sentinel

Browse Source
This commit is contained in:
Jason McVetta
2020-02-12 17:53:52 +07:00
parent ea30c1d43f
commit d22c916bfc
2 changed files with 14 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
django_q/cluster.py
+3 -2
View File
@@ -50,7 +50,7 @@ class Cluster(object):
self.stop_event = Event() self.stop_event = Event()
self.start_event = Event() self.start_event = Event()
self.sentinel = Process(target=Sentinel, 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() self.sentinel.start()
logger.info(_('Q Cluster-{} starting.').format(self.pid)) logger.info(_('Q Cluster-{} starting.').format(self.pid))
while not self.start_event.is_set(): while not self.start_event.is_set():
@@ -97,11 +97,12 @@ class Cluster(object):
class Sentinel(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 # Make sure we catch signals for the pool
signal.signal(signal.SIGINT, signal.SIG_IGN) signal.signal(signal.SIGINT, signal.SIG_IGN)
signal.signal(signal.SIGTERM, signal.SIG_DFL) signal.signal(signal.SIGTERM, signal.SIG_DFL)
self.pid = current_process().pid self.pid = current_process().pid
self.cluster_id = cluster_id
self.parent_pid = get_ppid() self.parent_pid = get_ppid()
self.name = current_process().name self.name = current_process().name
self.broker = broker or get_broker() self.broker = broker or get_broker()
django_q/tests/test_cluster.py
+11 -5
View File
@@ -3,6 +3,7 @@ import threading
from multiprocessing import Event, Value from multiprocessing import Event, Value
from time import sleep from time import sleep
from django.utils import timezone from django.utils import timezone
import uuid as uuidlib
import os import os
import pytest import pytest
@@ -72,7 +73,8 @@ def test_sentinel():
start_event = Event() start_event = Event()
stop_event = Event() stop_event = Event()
stop_event.set() 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 start_event.is_set()
assert s.status() == Conf.STOPPED 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) async_task('time.sleep', 5, broker=broker, **async_task_kwargs)
start_event = Event() start_event = Event()
stop_event = Event() stop_event = Event()
cluster_id = uuidlib.uuid4()
# Set a timer to stop the Sentinel # Set a timer to stop the Sentinel
threading.Timer(3, stop_event.set).start() 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 start_event.is_set()
assert s.status() == Conf.STOPPED assert s.status() == Conf.STOPPED
assert s.reincarnations == 1 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) async_task('time.sleep', 3, broker=broker, **async_task_kwargs)
start_event = Event() start_event = Event()
stop_event = Event() stop_event = Event()
cluster_id = uuidlib.uuid4()
# Set a timer to stop the Sentinel # Set a timer to stop the Sentinel
threading.Timer(6, stop_event.set).start() 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 start_event.is_set()
assert s.status() == Conf.STOPPED assert s.status() == Conf.STOPPED
assert s.reincarnations == 0 assert s.reincarnations == 0
@@ -297,12 +301,13 @@ def test_recycle(broker, monkeypatch):
async_task('django_q.tests.tasks.multiply', 2, 2, broker=broker) async_task('django_q.tests.tasks.multiply', 2, 2, broker=broker)
start_event = Event() start_event = Event()
stop_event = Event() stop_event = Event()
cluster_id = uuidlib.uuid4()
# override settings # override settings
monkeypatch.setattr(Conf, 'RECYCLE', 2) monkeypatch.setattr(Conf, 'RECYCLE', 2)
monkeypatch.setattr(Conf, 'WORKERS', 1) monkeypatch.setattr(Conf, 'WORKERS', 1)
# set a timer to stop the Sentinel # set a timer to stop the Sentinel
threading.Timer(3, stop_event.set).start() 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 start_event.is_set()
assert s.status() == Conf.STOPPED assert s.status() == Conf.STOPPED
assert s.reincarnations == 1 assert s.reincarnations == 1
@@ -333,7 +338,8 @@ def test_bad_secret(broker, monkeypatch):
stop_event = Event() stop_event = Event()
stop_event.set() stop_event.set()
start_event = Event() 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() Stat(s).save()
# change the SECRET # change the SECRET
monkeypatch.setattr(Conf, "SECRET_KEY", "OOPS") monkeypatch.setattr(Conf, "SECRET_KEY", "OOPS")