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

Add meaningfull process titles with currently running task name (#57)

Browse Source 
* Customize process names (in ps/top) and add task name in logs

* Increase severity of reincarnate log messages and add task name

* Document setproctitle optional dependency

Co-authored-by: Marc Sabatier <37879561+msabatier@users.noreply.github.com>
Co-authored-by: Stan Triepels <1939656+GDay@users.noreply.github.com>
This commit is contained in:
msabatier
2023-01-26 02:21:56 +01:00
committed by GitHub
co-authored by Stan Triepels
parent 650c3b1524
commit 17c1609f10
8 changed files with 1399 additions and 1242 deletions
Download Patch File Download Diff File Expand all files Collapse all files
django_q/cluster.py
+57 -18
View File
@@ -33,6 +33,7 @@ from django_q.conf import (
get_ppid,
logger,
psutil,
setproctitle,
resource,
)
from django_q.humanhash import humanize
@@ -59,6 +60,8 @@ class Cluster:
signal.signal(signal.SIGINT, self.sig_handler)
def start(self) -> int:
if setproctitle:
setproctitle.setproctitle(f"qcluster {current_process().name} {self.name}")
# Start Sentinel
self.stop_event = Event()
self.start_event = Event()
@@ -217,13 +220,13 @@ class Sentinel:
db.connections.close_all()
if process == self.monitor:
self.monitor = self.spawn_monitor()
logger.error(
logger.critical(
_("reincarnated monitor %(name)s after sudden death")
% {"name": process.name}
)
elif process == self.pusher:
self.pusher = self.spawn_pusher()
logger.error(
logger.critical(
_("reincarnated pusher %(name)s after sudden death")
% {"name": process.name}
)
@@ -233,15 +236,30 @@ class Sentinel:
if process.timer.value == 0:
# only need to terminate on timeout, otherwise we risk destabilizing
# the queues
task_name = ""
if psutil:
try:
process_name = psutil.Process(process.pid).name()
name_splits = process_name.split(" ")
task_name = name_splits[3] if len(name_splits) >= 4 and name_splits[2] == "processing" else ""
except psutil.NoSuchProcess:
pass
process.terminate()
logger.warning(
_("reincarnated worker %(name)s after timeout")
% {"name": process.name}
)
if task_name:
msg = (
_("reincarnated worker %(name)s after timeout while processing task %(task_name)s")
% {"name": process.name, "task_name": task_name}
)
else:
msg = (
_("reincarnated worker %(name)s after timeout")
% {"name": process.name}
)
logger.critical(msg)
elif int(process.timer.value) == -2:
logger.info(_("recycled worker %(name)s") % {"name": process.name})
else:
logger.error(
logger.critical(
_("reincarnated worker %(name)s after death")
% {"name": process.name}
)
@@ -358,9 +376,12 @@ def pusher(task_queue: Queue, event: Event, broker: Broker = None):
"""
if not broker:
broker = get_broker()
proc_name = current_process().name
if setproctitle:
setproctitle.setproctitle(f"qcluster {proc_name} pusher")
logger.info(
_("%(process_name)s pushing tasks at %(id)s")
% {"process_name": current_process().name, "id": current_process().pid}
_("%(name)s pushing tasks at %(id)s")
% {"name": proc_name, "id": current_process().pid}
)
while True:
try:
@@ -398,9 +419,11 @@ def monitor(result_queue: Queue, broker: Broker = None):
"""
if not broker:
broker = get_broker()
name = current_process().name
proc_name = current_process().name
if setproctitle:
setproctitle.setproctitle(f"qcluster {proc_name} monitor")
logger.info(
_("%(name)s monitoring at %(id)s") % {"name": name, "id": current_process().pid}
_("%(name)s monitoring at %(id)s") % {"name": proc_name, "id": current_process().pid}
)
for task in iter(result_queue.get, "STOP"):
# save the result
@@ -432,7 +455,7 @@ def monitor(result_queue: Queue, broker: Broker = None):
"task_result": task["result"],
}
)
logger.info(_("%(name)s stopped monitoring results") % {"name": name})
logger.info(_("%(name)s stopped monitoring results") % {"name": proc_name})
def worker(
@@ -451,6 +474,8 @@ def worker(
_("%(proc_name)s ready for work at %(id)s")
% {"proc_name": proc_name, "id": current_process().pid}
)
if setproctitle:
setproctitle.setproctitle(f"qcluster {proc_name} idle")
task_count = 0
if timeout is None:
timeout = -1
@@ -459,18 +484,30 @@ def worker(
result = None
timer.value = -1 # Idle
task_count += 1
f = task["func"]
# Log task creation and set process name
# Get the function from the task
func = task["func"]
func_name = get_func_repr(func)
logger.info(
_("%(proc_name)s processing '%(func_name)s' (%(task_name)s)")
func_name = get_func_repr(f)
task_name = task["name"]
task_desc = (
_("%(proc_name)s processing %(task_name)s '%(func_name)s'")
% {
"proc_name": proc_name,
"func_name": func_name,
"task_name": task["name"],
"task_name": task_name,
}
)
f = task["func"]
if "group" in task:
task_desc += f" [{task['group']}]"
logger.info(task_desc)
if setproctitle:
proc_title = f"qcluster {proc_name} processing {task_name} '{func_name}'"
if "group" in task:
proc_title += f" [{task['group']}]"
setproctitle.setproctitle(proc_title)
# if it's not an instance try to get it from the string
if not callable(f):
# locate() returns None if f cannot be loaded
@@ -500,6 +537,8 @@ def worker(
task["stopped"] = timezone.now()
result_queue.put(task)
timer.value = -1 # Idle
if setproctitle:
setproctitle.setproctitle(f"qcluster {proc_name} idle")
# Recycle
if task_count == Conf.RECYCLE or rss_check():
timer.value = -2 # Recycled