Feature/add scheduler pid info

django_rq/models.py

@@ -13,7 +13,6 @@
 class Queue(models.Model):
  """Placeholder model with no database table, but with django admin page
  and contenttype permission"""
-
  class Meta:
  managed = False # not in Django's database
  default_permissions = ()

django_rq/templates/django_rq/stats.html

@@ -36,6 +36,9 @@
  <th>Host</th>
  <th>Port</th>
  <th>DB</th>
+ {% if queue.scheduler_pid is not False %}
+ <th>Scheduler PID</th>
+ {% endif%}
  </tr>
  </thead>
  <tbody>
@@ -84,6 +87,9 @@
  <td>{{ queue.connection_kwargs.host }}</td>
  <td>{{ queue.connection_kwargs.port }}</td>
  <td>{{ queue.connection_kwargs.db }}</td>
+ {% if queue.scheduler_pid is not False %}
+ <td>{{ queue.scheduler_pid|default_if_none:"Inactive" }}</td>
+ {% endif %}
  </tr>
  {% endfor %}
  </tbody>

django_rq/tests/settings.py

@@ -148,6 +148,30 @@
  'PORT': 6379,
  'DB': 0,
  },
+ 'scheduler_scheduler_active_test': {
+ 'HOST': REDIS_HOST,
+ 'PORT': 6379,
+ 'DB': 0,
+ 'ASYNC': False,
+ },
+ 'scheduler_scheduler_inactive_test': {
+ 'HOST': REDIS_HOST,
+ 'PORT': 6379,
+ 'DB': 0,
+ 'ASYNC': False,
+ },
+ 'worker_scheduler_active_test': {
+ 'HOST': REDIS_HOST,
+ 'PORT': 6379,
+ 'DB': 0,
+ 'ASYNC': False,
+ },
+ 'worker_scheduler_inactive_test': {
+ 'HOST': REDIS_HOST,
+ 'PORT': 6379,
+ 'DB': 0,
+ 'ASYNC': False,
+ },
  'django-redis': {
  'USE_REDIS_CACHE': 'default',
  },

django_rq/tests/tests.py

@@ -32,7 +32,7 @@
 from django_rq import thread_queue
 from django_rq.templatetags.django_rq import force_escape, to_localtime
 from django_rq.tests.fixtures import DummyJob, DummyQueue, DummyWorker
-from django_rq.utils import get_jobs, get_statistics
+from django_rq.utils import get_jobs, get_statistics, get_scheduler_pid
 from django_rq.workers import get_worker, get_worker_class
 
 try:
@@ -808,6 +808,86 @@ def test_force_escape_regular_string(self):
  self.assertEqual(escaped_string, expected)
 
 
+class SchedulerPIDTest(TestCase):
+ @skipIf(RQ_SCHEDULER_INSTALLED is False, 'RQ Scheduler not installed')
+ def test_scheduler_scheduler_pid_active(self):
+ test_queue = 'scheduler_scheduler_active_test'
+ queues = [{
+ 'connection_config': {
+ 'DB': 0,
+ 'HOST': 'localhost',
+ 'PORT': 6379,
+ },
+ 'name': test_queue,
+ }] 
+ with patch('django_rq.utils.QUEUES_LIST',
+ new_callable=PropertyMock(return_value=queues)):
+ scheduler = get_scheduler(test_queue)
+ scheduler.register_birth()
+ self.assertIs(get_scheduler_pid(get_queue(scheduler.queue_name)), False)
+ scheduler.register_death()
+
+ @skipIf(RQ_SCHEDULER_INSTALLED is False, 'RQ Scheduler not installed')
+ def test_scheduler_scheduler_pid_inactive(self):
+ test_queue = 'scheduler_scheduler_inactive_test'
+ queues = [{
+ 'connection_config': {
+ 'DB': 0,
+ 'HOST': 'localhost',
+ 'PORT': 6379,
+ },
+ 'name': test_queue,
+ }] 
+ with patch('django_rq.utils.QUEUES_LIST',
+ new_callable=PropertyMock(return_value=queues)):
+ connection = get_connection(test_queue)
+ connection.flushall() # flush is needed to isolate from other tests
+ scheduler = get_scheduler(test_queue)
+ scheduler.remove_lock()
+ scheduler.register_death() # will mark the scheduler as death so get_scheduler_pid will return None
+ self.assertIs(get_scheduler_pid(get_queue(scheduler.queue_name)), False)
+
+ @skipIf(RQ_SCHEDULER_INSTALLED is True, 'RQ Scheduler installed (no worker--with-scheduler)')
+ def test_worker_scheduler_pid_active(self):
+ '''The worker works as scheduler too if RQ Scheduler not installed, and the pid scheduler_pid is correct'''
+ test_queue = 'worker_scheduler_active_test'
+ queues = [{
+ 'connection_config': {
+ 'DB': 0,
+ 'HOST': 'localhost',
+ 'PORT': 6379,
+ },
+ 'name': test_queue,
+ }]
+ with patch('rq.scheduler.RQScheduler.release_locks') as mock_release_locks:
+ with patch('django_rq.utils.QUEUES_LIST',
+ new_callable=PropertyMock(return_value=queues)):
+ queue = get_queue(test_queue)
+ worker = get_worker(test_queue, name=uuid4().hex)
+ worker.work(with_scheduler=True, burst=True) # force the worker to acquire a scheduler lock
+ pid = get_scheduler_pid(queue)
+ self.assertIsNotNone(pid)
+ self.assertIsNot(pid, False)
+ self.assertIsInstance(pid, int)
+
+ @skipIf(RQ_SCHEDULER_INSTALLED is True, 'RQ Scheduler installed (no worker--with-scheduler)')
+ def test_worker_scheduler_pid_inactive(self):
+ '''The worker works as scheduler too if RQ Scheduler not installed, and the pid scheduler_pid is correct'''
+ test_queue = 'worker_scheduler_inactive_test'
+ queues = [{
+ 'connection_config': {
+ 'DB': 0,
+ 'HOST': 'localhost',
+ 'PORT': 6379,
+ },
+ 'name': test_queue,
+ }] 
+ with patch('django_rq.utils.QUEUES_LIST',
+ new_callable=PropertyMock(return_value=queues)):
+ worker = get_worker(test_queue, name=uuid4().hex)
+ worker.work(with_scheduler=False, burst=True) # worker will not acquire lock, scheduler_pid should return None
+ self.assertIsNone(get_scheduler_pid(worker.queues[0]))
+
 class UtilsTest(TestCase):
  def test_get_statistics(self):
  """get_statistics() returns the right number of workers"""

django_rq/utils.py

@@ -10,9 +10,32 @@
 from rq.worker import Worker
 from rq.worker_registration import clean_worker_registry
 
-from .queues import get_connection, get_queue_by_index
+
+from .queues import get_connection, get_queue_by_index, get_scheduler
 from .settings import QUEUES_LIST
 from .templatetags.django_rq import to_localtime
+from django.core.exceptions import ImproperlyConfigured
+
+def get_scheduler_pid(queue):
+ '''Checks whether there's a scheduler-lock on a particular queue, and returns the PID.
+ It Only works with RQ's Built-in RQScheduler.
+ When RQ-Scheduler is available returns False 
+ If not, it checks the RQ's RQScheduler for a scheduler lock in the desired queue
+ Note: result might have some delay (1-15 minutes) but it helps visualizing whether the setup is working correcly
+ '''
+ try:
+ # first try get the rq-scheduler
+ scheduler = get_scheduler(queue.name) # should fail if rq_scheduler not present
+ return False # Not possible to give useful information without creating a performance issue (redis.keys())
+ except ImproperlyConfigured:
+ from rq.scheduler import RQScheduler
+ # When a scheduler acquires a lock it adds an expiring key: (e.g: rq:scheduler-lock:<queue.name>)
+ #TODO: (RQ>= 1.13) return queue.scheduler_pid 
+ pid = queue.connection.get(RQScheduler.get_locking_key(queue.name))
+ return int(pid.decode()) if pid is not None else None
+ except Exception as e:
+ pass # Return None
+ return None
 
 
 def get_statistics(run_maintenance_tasks=False):
@@ -47,6 +70,7 @@ def get_statistics(run_maintenance_tasks=False):
  'oldest_job_timestamp': oldest_job_timestamp,
  'index': index,
  'connection_kwargs': connection_kwargs,
+ 'scheduler_pid': get_scheduler_pid(queue),
  }
 
  connection = get_connection(queue.name)