wip

Author: cunla

scheduler/views/queue_views.py

@@ -6,7 +6,7 @@
 from django.contrib.admin.views.decorators import staff_member_required
 from django.http import HttpResponse, HttpRequest, HttpResponseNotFound, JsonResponse
 from django.shortcuts import render, redirect
-from django.urls import reverse
+from django.urls import reverse, resolve
 from django.utils.http import url_has_allowed_host_and_scheme
 from django.views.decorators.cache import never_cache
 
@@ -105,10 +105,7 @@ def clear_queue_registry(request: HttpRequest, queue_name: str, registry_name: s
     registry = queue.get_registry(registry_name)
     if registry is None:
         return HttpResponseNotFound()
-
-    next_url = request.META.get("HTTP_REFERER") or reverse("queue_registry_jobs", args=[queue_name, registry_name])
-    if not url_has_allowed_host_and_scheme(next_url, allowed_hosts=None):
-        messages.warning(request, "Bad followup URL")
+    next_url = _check_next_url(request, reverse("queue_registry_jobs", args=[queue_name, registry_name]))
     if request.method == "POST":
         try:
             if registry is queue:
@@ -204,15 +201,23 @@ def queue_confirm_action(request: HttpRequest, queue_name: str) -> HttpResponse:
     return redirect(next_url)
 
 
+def _check_next_url(request: HttpRequest, default_next_url: str) -> str:
+    next_url = request.POST.get("next_url", default_next_url)
+    if not url_has_allowed_host_and_scheme(next_url, allowed_hosts=None):
+        messages.warning(request, "Bad followup URL")
+    try:
+        resolve(next_url)
+    except Exception:
+        messages.warning(request, "Bad followup URL")
+        next_url = default_next_url
+    return next_url
+
+
 @never_cache
 @staff_member_required
 def queue_actions(request: HttpRequest, queue_name: str) -> HttpResponse:
     queue = get_queue(queue_name)
-    next_url = request.POST.get("next_url") or reverse("queue_registry_jobs", args=[queue_name, "queued"])
-    if not url_has_allowed_host_and_scheme(next_url, allowed_hosts=None):
-        messages.warning(request, "Bad followup URL")
-        next_url = reverse("queue_registry_jobs", args=[queue_name, "queued"])
-
+    next_url = _check_next_url(request, reverse("queue_registry_jobs", args=[queue_name, "queued"]))
     action = request.POST.get("action", False)
     job_names = request.POST.get("job_names", False)
     if request.method != "POST" or not action or not job_names:

scheduler/worker/scheduler.py

@@ -5,6 +5,7 @@
 from datetime import datetime
 from enum import Enum
 from multiprocessing import Process
+from threading import Thread
 from typing import List, Set, Optional, Sequence, Dict
 
 import django
@@ -52,7 +53,7 @@ def __init__(
         self.interval = interval
         self._stop_requested = False
         self._status = SchedulerStatus.STOPPED
-        self._process = None
+        self._thread = None
         self._pid: Optional[int] = None
         self.worker_name = worker_name
 
@@ -96,8 +97,8 @@ def start(self, burst=False) -> None:
             self.release_locks()
             return
         self._status = SchedulerStatus.STARTED
-        self._process = Process(target=run_scheduler, args=(self,), name="Scheduler")
-        self._process.start()
+        self._thread = Thread(target=run_scheduler, args=(self,), name="scheduler-thread")
+        self._thread.start()
 
     def _install_signal_handlers(self):
         """Installs signal handlers for handling SIGINT and SIGTERM"""

scheduler/worker/worker.py

@@ -401,7 +401,7 @@ def dequeue_job_and_maintain_ttl(
 
         self._model.set_field("state", WorkerStatus.IDLE, connection=self.connection)
         self.procline(f"Listening on {qnames}")
-        logger.debug(f"*** Worker {self.name} listening on {qnames}...")
+        logger.debug(f"[Worker {self.name}/{self._pid}]: listening on {qnames}...")
         connection_wait_time = 1.0
         idle_since = utcnow()
         idle_time_left = max_idle_time
@@ -593,8 +593,8 @@ def stop_scheduler(self):
             os.kill(self.scheduler.pid, signal.SIGTERM)
         except OSError:
             pass
-        if self.scheduler._process is not None:
-            self.scheduler._process.join()
+        if self.scheduler._thread is not None:
+            self.scheduler._thread.join()
 
     def refresh(self, update_queues: bool = False):
         """Refreshes the worker data.