Add: periodic job to remove ghost locks. (getredash#5087)

redash/tasks/__init__.py

@@ -12,6 +12,7 @@
     refresh_schemas,
     cleanup_query_results,
     empty_schedules,
+    remove_ghost_locks,
 )
 from .alerts import check_alerts_for_query
 from .failure_report import send_aggregated_errors

redash/tasks/queries/__init__.py

@@ -3,5 +3,6 @@
     refresh_schemas,
     cleanup_query_results,
     empty_schedules,
+    remove_ghost_locks,
 )
 from .execution import execute_query, enqueue_query

redash/tasks/queries/maintenance.py

@@ -10,6 +10,7 @@
 from redash.tasks.failure_report import track_failure
 from redash.utils import json_dumps, sentry
 from redash.worker import job, get_job_logger
+from redash.monitor import rq_job_ids
 
 from .execution import enqueue_query
 
@@ -134,6 +135,24 @@ def cleanup_query_results():
     logger.info("Deleted %d unused query results.", deleted_count)
 
 
+def remove_ghost_locks():
+    """
+    Removes query locks that reference a non existing RQ job.
+    """
+    keys = redis_connection.keys("query_hash_job:*")
+    locks = {k: redis_connection.get(k) for k in keys}
+    jobs = list(rq_job_ids())
+
+    count = 0
+
+    for lock, job_id in locks.items():
+        if job_id not in jobs:
+            redis_connection.delete(lock)
+            count += 1
+
+    logger.info("Locks found: {}, Locks removed: {}".format(len(locks), count))
+
+
 @job("schemas")
 def refresh_schema(data_source_id):
     ds = models.DataSource.get_by_id(data_source_id)

redash/tasks/schedule.py

@@ -11,6 +11,7 @@
 from redash.tasks import (
     sync_user_details,
     refresh_queries,
+    remove_ghost_locks,
     empty_schedules,
     refresh_schemas,
     cleanup_query_results,
@@ -61,6 +62,11 @@ def schedule(kwargs):
 def periodic_job_definitions():
     jobs = [
         {"func": refresh_queries, "interval": 30, "result_ttl": 600},
+        {
+            "func": remove_ghost_locks,
+            "interval": timedelta(minutes=1),
+            "result_ttl": 600,
+        },
         {"func": empty_schedules, "interval": timedelta(minutes=60)},
         {
             "func": refresh_schemas,