PTFE-1419 ensure we are periodically indexing redis data (#533)

runner_manager/jobs/startup.py

@@ -43,20 +43,28 @@ def sync_runner_groups(settings: Settings):
         runner_group.delete(pk=runner_group.pk, github=github)
 
 
-def bootstrap_healthchecks(
+def bootstrap_scheduler(
     settings: Settings,
 ):
     scheduler: Scheduler = get_scheduler()
     jobs: List[Job] = scheduler.get_jobs()
     groups: List[RunnerGroup] = RunnerGroup.find().all()
     for job in jobs:
         # Cancel any existing healthcheck jobs
-        if job.meta.get("type") == "healthcheck":
-            log.info(
-                f"Canceling healthcheck job {job.id} for group {job.meta['group']}"
-            )
+        job_type = job.meta.get("type")
+        if job_type == "healthcheck" or job_type == "migrator":
+            log.info(f"Canceling {job_type} job: {job.id}")
             scheduler.cancel(job)
 
+    # Scheduling indexing job
+    scheduler.schedule(
+        scheduled_time=datetime.utcnow(),
+        func=indexing,
+        interval=settings.indexing_interval.total_seconds(),
+        meta={"type": "indexing"},
+        result_ttl=settings.indexing_interval.total_seconds() * 10,
+        repeat=None,
+    )
     for group in groups:
         log.info(f"Scheduling healthcheck for group {group.name}")
         scheduler.schedule(
@@ -81,14 +89,18 @@ def bootstrap_healthchecks(
         )
 
 
+def indexing():
+    Migrator().run()
+
+
 def startup(settings: Settings = get_settings()):
     """Bootstrap the application."""
     log.info("Startup initiated.")
-    Migrator().run()
+    indexing()
     log.info("Creating runner groups...")
     sync_runner_groups(settings)
     log.info("Runner groups created.")
     log.info("Bootstrapping healthchecks...")
-    bootstrap_healthchecks(settings)
+    bootstrap_scheduler(settings)
     log.info("Healthchecks bootstrapped.")
     log.info("Startup complete.")

runner_manager/models/settings.py

@@ -34,6 +34,7 @@ class Settings(BaseSettings):
     timeout_runner: timedelta = timedelta(minutes=15)
     time_to_live: Optional[timedelta] = timedelta(hours=12)
     healthcheck_interval: timedelta = timedelta(minutes=15)
+    indexing_interval: timedelta = timedelta(hours=1)
     github_base_url: Optional[AnyHttpUrl] = Field(default="https://api.github.com")
     github_webhook_secret: Optional[SecretStr] = None
     github_token: Optional[SecretStr] = None

tests/unit/jobs/test_startup.py

@@ -1,5 +1,8 @@
+from typing import List
+
 from rq import Queue
 from rq.job import Job, JobStatus
+from rq_scheduler import Scheduler
 
 from runner_manager import Runner, RunnerGroup, Settings
 from runner_manager.clients.github import GitHub
@@ -43,6 +46,25 @@ def test_startup(queue: Queue, settings: Settings, github: GitHub):
     assert Runner.find().count() == runners - 1
 
 
+def test_scheduler(
+    queue: Queue, settings: Settings, github: GitHub, scheduler: Scheduler
+):
+    """Test that the scheduler is bootstrapped with the correct jobs."""
+    queue.enqueue(startup, settings)
+    jobs: List[Job] = scheduler.get_jobs()
+    is_indexing: bool = False
+    is_healthcheck: bool = False
+    for job in jobs:
+        job_type = job.meta.get("type")
+        if job_type == "indexing":
+            is_indexing = True
+        elif job_type == "healthcheck":
+            is_healthcheck = True
+
+    assert is_indexing is True
+    assert is_healthcheck is True
+
+
 def test_update_group_sync(settings: Settings, github: GitHub):
     sync_runner_groups(settings)
     runner_group: RunnerGroup = RunnerGroup.find().first()