mirror of
https://github.com/django-q2/django-q2.git
synced 2026-09-30 04:08:11 +08:00
Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5ff834b794 | ||
|
|
2fe20033e3 | ||
|
|
1537872bf2 | ||
|
|
c9020ce94e | ||
|
|
5597d7ccd8 | ||
|
|
480b77dfe4 |
@@ -27,34 +27,19 @@ jobs:
|
|||||||
strategy:
|
strategy:
|
||||||
matrix:
|
matrix:
|
||||||
python-version:
|
python-version:
|
||||||
- "3.9"
|
|
||||||
- "3.10"
|
- "3.10"
|
||||||
- "3.11"
|
- "3.11"
|
||||||
- "3.12"
|
- "3.12"
|
||||||
|
- "3.13"
|
||||||
|
- "3.14"
|
||||||
django:
|
django:
|
||||||
- "4.2"
|
|
||||||
- "5.0"
|
|
||||||
- "5.1"
|
|
||||||
- "5.2"
|
- "5.2"
|
||||||
- "6.0"
|
- "6.0"
|
||||||
exclude:
|
exclude:
|
||||||
# django 5.2 does not support 3.9
|
# django 6 does not support 3.10
|
||||||
- python-version: "3.9"
|
|
||||||
django: "5.2"
|
|
||||||
# django 5.1 does not support 3.9
|
|
||||||
- python-version: "3.9"
|
|
||||||
django: "5.1"
|
|
||||||
# django 5.0 does not support 3.9
|
|
||||||
- python-version: "3.9"
|
|
||||||
django: "5.0"
|
|
||||||
# django 4.2 does not support 3.13
|
|
||||||
- python-version: "3.13"
|
|
||||||
django: "4.2"
|
|
||||||
# django 6.0 does not support earlier than 3.12
|
|
||||||
- python-version: "3.9"
|
|
||||||
django: "6.0"
|
|
||||||
- python-version: "3.10"
|
- python-version: "3.10"
|
||||||
django: "6.0"
|
django: "6.0"
|
||||||
|
# django 6 does not support 3.11
|
||||||
- python-version: "3.11"
|
- python-version: "3.11"
|
||||||
django: "6.0"
|
django: "6.0"
|
||||||
|
|
||||||
|
|||||||
+1
-1
@@ -1,5 +1,5 @@
|
|||||||
# Sets the python version
|
# Sets the python version
|
||||||
FROM python:3.9-slim-bookworm
|
FROM python:3.12-slim-bookworm
|
||||||
|
|
||||||
# Allows the logs generated by python apps to be rendered in the terminal
|
# Allows the logs generated by python apps to be rendered in the terminal
|
||||||
ENV PYTHONUNBUFFERED 1
|
ENV PYTHONUNBUFFERED 1
|
||||||
|
|||||||
+3
-3
@@ -35,13 +35,13 @@ See the `changelog <https://github.com/GDay/django-q2/blob/master/CHANGELOG.md>`
|
|||||||
Requirements
|
Requirements
|
||||||
~~~~~~~~~~~~
|
~~~~~~~~~~~~
|
||||||
|
|
||||||
- `Django <https://www.djangoproject.com>`__ > = 4.2
|
- `Django <https://www.djangoproject.com>`__ 5.2 and 6.0
|
||||||
- `Django-picklefield <https://github.com/gintas/django-picklefield>`__
|
- `Django-picklefield <https://github.com/gintas/django-picklefield>`__
|
||||||
|
|
||||||
Tested with:
|
Tested with:
|
||||||
|
|
||||||
* Python 3.9 to 3.13.
|
* Python 3.10 to 3.14.
|
||||||
* Django 4.2 to 6.0.
|
* Django 5.2 and 6.0.
|
||||||
|
|
||||||
Brokers
|
Brokers
|
||||||
~~~~~~~
|
~~~~~~~
|
||||||
|
|||||||
+27
-3
@@ -73,7 +73,30 @@ class Cluster:
|
|||||||
)
|
)
|
||||||
self.sentinel.start()
|
self.sentinel.start()
|
||||||
logger.info(_("Q Cluster %(name)s starting.") % {"name": self.name})
|
logger.info(_("Q Cluster %(name)s starting.") % {"name": self.name})
|
||||||
while not self.start_event.is_set():
|
# Wait for the sentinel to set start_event by polling the event's state. We
|
||||||
|
# cannot call wait() as that will deadlock if a signal is received while blocked
|
||||||
|
# in wait(), as the sentinel will block in start_event.set() while this process
|
||||||
|
# is blocked in sentinel.join(). It is also necessary to check for start_event
|
||||||
|
# being set to None by the signal handler.
|
||||||
|
while self.start_event and not self.start_event.is_set():
|
||||||
|
# While waiting for the sentinel to start, also check for sentinel premature
|
||||||
|
# death
|
||||||
|
if not self.sentinel.is_alive():
|
||||||
|
if self.sentinel.exitcode == 0:
|
||||||
|
# the cluster was stopped via SIGTERM/SIGINT and sentinel early
|
||||||
|
# termination was intentional -- exit quietly
|
||||||
|
break
|
||||||
|
logger.error(
|
||||||
|
_(
|
||||||
|
"Q Cluster %(name)s failed to start. Sentinel exited "
|
||||||
|
"with exit code %(exitcode)s."
|
||||||
|
)
|
||||||
|
% {"name": self.name, "exitcode": self.sentinel.exitcode}
|
||||||
|
)
|
||||||
|
raise RuntimeError(
|
||||||
|
f"Q Cluster {self.name} failed to start: sentinel exited "
|
||||||
|
f"with code {self.sentinel.exitcode}"
|
||||||
|
)
|
||||||
sleep(0.1)
|
sleep(0.1)
|
||||||
return self.pid
|
return self.pid
|
||||||
|
|
||||||
@@ -313,7 +336,7 @@ class Sentinel:
|
|||||||
counter = 0
|
counter = 0
|
||||||
cycle = Conf.GUARD_CYCLE # guard loop sleep in seconds
|
cycle = Conf.GUARD_CYCLE # guard loop sleep in seconds
|
||||||
# Guard loop. Runs at least once
|
# Guard loop. Runs at least once
|
||||||
while not self.stop_event.is_set() or not counter:
|
while True:
|
||||||
# Check Workers
|
# Check Workers
|
||||||
for p in self.pool:
|
for p in self.pool:
|
||||||
with p.timer.get_lock():
|
with p.timer.get_lock():
|
||||||
@@ -337,7 +360,8 @@ class Sentinel:
|
|||||||
scheduler(broker=self.broker)
|
scheduler(broker=self.broker)
|
||||||
# Save current status
|
# Save current status
|
||||||
Stat(self).save()
|
Stat(self).save()
|
||||||
sleep(cycle)
|
if self.stop_event.wait(cycle):
|
||||||
|
break
|
||||||
self.stop()
|
self.stop()
|
||||||
|
|
||||||
def stop(self):
|
def stop(self):
|
||||||
|
|||||||
@@ -97,6 +97,8 @@ def monitor(run_once=False, broker=None):
|
|||||||
for stat in stats:
|
for stat in stats:
|
||||||
status = stat.status
|
status = stat.status
|
||||||
# color status
|
# color status
|
||||||
|
if stat.status == Conf.STARTING:
|
||||||
|
status = term.yellow(str(Conf.STARTING))
|
||||||
if stat.status == Conf.WORKING:
|
if stat.status == Conf.WORKING:
|
||||||
status = term.green(str(Conf.WORKING))
|
status = term.green(str(Conf.WORKING))
|
||||||
elif stat.status == Conf.STOPPING:
|
elif stat.status == Conf.STOPPING:
|
||||||
@@ -105,6 +107,8 @@ def monitor(run_once=False, broker=None):
|
|||||||
status = term.red(str(Conf.STOPPED))
|
status = term.red(str(Conf.STOPPED))
|
||||||
elif stat.status == Conf.IDLE:
|
elif stat.status == Conf.IDLE:
|
||||||
status = str(Conf.IDLE)
|
status = str(Conf.IDLE)
|
||||||
|
else:
|
||||||
|
status = _("Unknown")
|
||||||
# color q's
|
# color q's
|
||||||
tasks = str(stat.task_q_size)
|
tasks = str(stat.task_q_size)
|
||||||
if stat.task_q_size > 0:
|
if stat.task_q_size > 0:
|
||||||
@@ -138,7 +142,7 @@ def monitor(run_once=False, broker=None):
|
|||||||
)
|
)
|
||||||
print(
|
print(
|
||||||
term.move(i, 3 * col_width)
|
term.move(i, 3 * col_width)
|
||||||
+ term.center(workers, width=col_width - 1)
|
+ term.center(str(workers), width=col_width - 1)
|
||||||
)
|
)
|
||||||
print(
|
print(
|
||||||
term.move(i, 4 * col_width)
|
term.move(i, 4 * col_width)
|
||||||
@@ -146,11 +150,11 @@ def monitor(run_once=False, broker=None):
|
|||||||
)
|
)
|
||||||
print(
|
print(
|
||||||
term.move(i, 5 * col_width)
|
term.move(i, 5 * col_width)
|
||||||
+ term.center(results, width=col_width - 1)
|
+ term.center(str(results), width=col_width - 1)
|
||||||
)
|
)
|
||||||
print(
|
print(
|
||||||
term.move(i, 6 * col_width)
|
term.move(i, 6 * col_width)
|
||||||
+ term.center(stat.reincarnations, width=col_width - 1)
|
+ term.center(str(stat.reincarnations), width=col_width - 1)
|
||||||
)
|
)
|
||||||
print(
|
print(
|
||||||
term.move(i, 7 * col_width)
|
term.move(i, 7 * col_width)
|
||||||
@@ -411,27 +415,27 @@ def memory(run_once=False, workers=False, broker=None):
|
|||||||
)
|
)
|
||||||
print(
|
print(
|
||||||
term.move(row, 2 * col_width)
|
term.move(row, 2 * col_width)
|
||||||
+ term.center(memory_available_percentage, width=col_width - 1)
|
+ term.center(str(memory_available_percentage), width=col_width - 1)
|
||||||
)
|
)
|
||||||
print(
|
print(
|
||||||
term.move(row, 3 * col_width)
|
term.move(row, 3 * col_width)
|
||||||
+ term.center(memory_available, width=col_width - 1)
|
+ term.center(str(memory_available), width=col_width - 1)
|
||||||
)
|
)
|
||||||
print(
|
print(
|
||||||
term.move(row, 4 * col_width)
|
term.move(row, 4 * col_width)
|
||||||
+ term.center(
|
+ term.center(
|
||||||
round(psutil.virtual_memory().total / 1024**2, 2),
|
str(round(psutil.virtual_memory().total / 1024**2, 2)),
|
||||||
width=col_width - 1,
|
width=col_width - 1,
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
print(
|
print(
|
||||||
term.move(row, 5 * col_width)
|
term.move(row, 5 * col_width)
|
||||||
+ term.center(get_process_mb(stat.sentinel), width=col_width - 1)
|
+ term.center(str(get_process_mb(stat.sentinel)), width=col_width - 1)
|
||||||
)
|
)
|
||||||
print(
|
print(
|
||||||
term.move(row, 6 * col_width)
|
term.move(row, 6 * col_width)
|
||||||
+ term.center(
|
+ term.center(
|
||||||
get_process_mb(getattr(stat, "monitor", None)),
|
str(get_process_mb(getattr(stat, "monitor", None))),
|
||||||
width=col_width - 1,
|
width=col_width - 1,
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
@@ -444,7 +448,7 @@ def memory(run_once=False, workers=False, broker=None):
|
|||||||
print(
|
print(
|
||||||
term.move(row, 7 * col_width)
|
term.move(row, 7 * col_width)
|
||||||
+ term.center(
|
+ term.center(
|
||||||
workers_mb or "NO_PROCESSES_FOUND", width=col_width - 1
|
str(workers_mb) or "NO_PROCESSES_FOUND", width=col_width - 1
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
row += 1
|
row += 1
|
||||||
@@ -476,7 +480,7 @@ def memory(run_once=False, workers=False, broker=None):
|
|||||||
mb_used = get_process_mb(worker_pid)
|
mb_used = get_process_mb(worker_pid)
|
||||||
print(
|
print(
|
||||||
term.move(row, (idx + 1) * col_width)
|
term.move(row, (idx + 1) * col_width)
|
||||||
+ term.center(mb_used, width=col_width - 1)
|
+ term.center(str(mb_used), width=col_width - 1)
|
||||||
)
|
)
|
||||||
row += 1
|
row += 1
|
||||||
row += 1
|
row += 1
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
import os
|
import os
|
||||||
|
import signal
|
||||||
import sys
|
import sys
|
||||||
import threading
|
import threading
|
||||||
import uuid as uuidlib
|
import uuid as uuidlib
|
||||||
@@ -98,6 +99,116 @@ def test_cluster_initial(broker):
|
|||||||
broker.delete_queue()
|
broker.delete_queue()
|
||||||
|
|
||||||
|
|
||||||
|
class TestEarlyClusterStop:
|
||||||
|
sentinel_event = Event()
|
||||||
|
test_event = Event()
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _fake_ping_for_test_cluster_early_stop():
|
||||||
|
"""
|
||||||
|
Fake broker ping() method for test_cluster_early_stop()
|
||||||
|
|
||||||
|
This is called in the sentinel process's start() method and overriding it allows
|
||||||
|
us to synchronize the sentinel process with the main process to provide a
|
||||||
|
deterministic ordering of operations for the test.
|
||||||
|
|
||||||
|
This is implemented as a static method so the patched broker remains pickleable
|
||||||
|
for passing to the sentinel process, in case anyone ever runs the test on
|
||||||
|
platforms like MacOS that use spawn Process start method.
|
||||||
|
"""
|
||||||
|
# let the main process know the sentinel is waiting
|
||||||
|
TestEarlyClusterStop.sentinel_event.set()
|
||||||
|
# wait for the main process to let the sentinel proceed. The timeout prevents
|
||||||
|
# the test from hanging if the main process terminates unexpectedly before
|
||||||
|
# setting the event.
|
||||||
|
assert TestEarlyClusterStop.test_event.wait(10)
|
||||||
|
|
||||||
|
@pytest.mark.django_db
|
||||||
|
def test_cluster_early_stop(self, broker, monkeypatch):
|
||||||
|
"""
|
||||||
|
Test stopping the cluster before the sentinel has set the cluster's start_event
|
||||||
|
"""
|
||||||
|
|
||||||
|
def raise_sigterm():
|
||||||
|
# wait for the sentinel to be blocked in fake_ping(). Periodically check
|
||||||
|
# that the sentinel is alive in case the sentinel dies unexpectedly before
|
||||||
|
# setting the sentinel_event -- prevents the test from hanging if sentinel
|
||||||
|
# dies unexpectedly.
|
||||||
|
while True:
|
||||||
|
if self.sentinel_event.wait(0.5):
|
||||||
|
break
|
||||||
|
if not c.sentinel.is_alive():
|
||||||
|
break
|
||||||
|
# stop the cluster via SIGTERM
|
||||||
|
os.kill(os.getpid(), signal.SIGTERM)
|
||||||
|
# unblock the sentinel
|
||||||
|
self.test_event.set()
|
||||||
|
|
||||||
|
monkeypatch.setattr(
|
||||||
|
broker,
|
||||||
|
"ping",
|
||||||
|
self._fake_ping_for_test_cluster_early_stop,
|
||||||
|
)
|
||||||
|
|
||||||
|
broker.list_key = "initial_test:q"
|
||||||
|
broker.delete_queue()
|
||||||
|
c = Cluster(broker=broker)
|
||||||
|
assert c.sentinel is None
|
||||||
|
assert c.stat.status == Conf.STOPPED
|
||||||
|
|
||||||
|
# Use a timer to stop the cluster while it is still starting up. The timer function
|
||||||
|
# is guaranteed to run before c.start() returns, as start() must wait for the
|
||||||
|
# sentinel to set the cluster's start_event, and the sentinel will block on
|
||||||
|
# test_event before setting start_event.
|
||||||
|
threading.Timer(0.5, raise_sigterm).start()
|
||||||
|
|
||||||
|
c.start()
|
||||||
|
stat = c.stat
|
||||||
|
assert stat.status == Conf.STOPPED
|
||||||
|
assert c.sentinel.is_alive() is False
|
||||||
|
assert c.has_stopped
|
||||||
|
broker.delete_queue()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.django_db
|
||||||
|
def test_cluster_stop_responsive(broker, monkeypatch):
|
||||||
|
"""
|
||||||
|
Ensure that stopping the cluster is responsive and does not wait for a full
|
||||||
|
GUARD_CYCLE.
|
||||||
|
"""
|
||||||
|
timeout_seconds = 5
|
||||||
|
|
||||||
|
def timeout(signal, frame):
|
||||||
|
assert 0, f"cluster did not stop after {timeout_seconds}s"
|
||||||
|
|
||||||
|
broker.list_key = "initial_test:q"
|
||||||
|
broker.delete_queue()
|
||||||
|
# set a long GUARD_CYCLE -- greater than timeout_seconds
|
||||||
|
monkeypatch.setattr(Conf, "GUARD_CYCLE", timeout_seconds * 2)
|
||||||
|
c = Cluster(broker=broker)
|
||||||
|
assert c.sentinel is None
|
||||||
|
assert c.stat.status == Conf.STOPPED
|
||||||
|
assert c.start() > 0
|
||||||
|
assert c.sentinel.is_alive() is True
|
||||||
|
assert c.is_running
|
||||||
|
assert c.is_stopping is False
|
||||||
|
assert c.is_starting is False
|
||||||
|
sleep(0.5)
|
||||||
|
stat = c.stat
|
||||||
|
assert stat.status == Conf.IDLE
|
||||||
|
prev_handler = signal.signal(signal.SIGALRM, timeout)
|
||||||
|
try:
|
||||||
|
signal.alarm(timeout_seconds)
|
||||||
|
assert c.stop() is True
|
||||||
|
finally:
|
||||||
|
signal.alarm(0) # cancel the alarm
|
||||||
|
signal.signal(signal.SIGALRM, prev_handler)
|
||||||
|
assert c.sentinel.is_alive() is False
|
||||||
|
assert c.has_stopped
|
||||||
|
assert c.stop() is False
|
||||||
|
broker.delete_queue()
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.django_db
|
@pytest.mark.django_db
|
||||||
def test_sentinel():
|
def test_sentinel():
|
||||||
start_event = Event()
|
start_event = Event()
|
||||||
|
|||||||
Generated
+853
-794
File diff suppressed because it is too large
Load Diff
+2
-2
@@ -52,8 +52,8 @@ include = [
|
|||||||
|
|
||||||
|
|
||||||
[tool.poetry.dependencies]
|
[tool.poetry.dependencies]
|
||||||
python = ">=3.9,<4"
|
python = ">=3.10,<4"
|
||||||
django = ">=4.2"
|
django = ">=5.2"
|
||||||
django-picklefield = "^3.1"
|
django-picklefield = "^3.1"
|
||||||
|
|
||||||
blessed = { version = "^1.19.1", optional = true }
|
blessed = { version = "^1.19.1", optional = true }
|
||||||
|
|||||||
Reference in New Issue
Block a user