WIP: Feat: Add celery beat for scheduling new cronjobs

README.md

@@ -190,9 +190,14 @@ For updating the index
 ```(bash)
 docker-compose exec serve bash python manage.py update_index
 ```
+## Cronjobs
+
+For more info checkout [Cronjobs](./docs/cronjobs.md)
+
 ## Sentry
 
-For updating the cron monitored tasks
+For updating the cron monitored tasks (legacy k8s CronJobs only — celery beat
+cronjobs are registered automatically)
 ```(bash)
 docker-compose exec serve bash ./manage.py cron_job_monitor
 ```

api/management/commands/run_celery_dev.py

@@ -4,10 +4,7 @@
 from django.core.management.base import BaseCommand
 from django.utils.autoreload import run_with_reloader
 
-from main.celery import Queues
-
-all_queues = ",".join([q for q in Queues.DEV_QUEUES])
-CMD = f"celery -A main worker -Q {all_queues} --concurrency=2 -l info"
+CMD = "celery -A main worker -E --concurrency=2 -l info"
 
 
 def restart_celery():

api/management/commands/run_celery_prod.py

@@ -5,9 +5,9 @@
 
 from django.core.management.base import BaseCommand
 
-from main.celery import Queues
+from main.cronjobs import CeleryQueue
 
-all_queues = ",".join([q for q in Queues.DEV_QUEUES])
+all_queues = ",".join([q.name for q in CeleryQueue.ALL_QUEUE])
 
 # NOTE: Use a fixed concurrency to prevent the pod from being OOMKilled,
 # as Celery defaults to one worker per available CPU.

api/tasks.py

@@ -2,10 +2,12 @@
 
 from celery import shared_task
 from django.contrib.auth.models import User
+from django.core import management
 from django.utils import timezone
 from rest_framework.authtoken.models import Token
 
 from api.playwright import render_pdf_from_url
+from main.lock import RedisLockKey, redis_lock
 from main.utils import logger_context
 
 from .logger import logger
@@ -59,3 +61,20 @@ def generate_export_pdf(export_id, title, set_user_language="en"):
         export.status = Export.ExportStatus.ERRORED
         export.save(update_fields=["status"])
     logger.info(f"End export: {export.pk}")
+
+
+# TODO(susilnem): Do we need this cron?
+@shared_task
+def clear_expired_django_sessions():
+    """Purge expired django_session rows -- nothing else prunes them."""
+    with redis_lock(RedisLockKey.CLEAR_EXPIRED_DJANGO_SESSIONS) as acquired:
+        if not acquired:
+            logger.warning("clear_expired_django_sessions: already running, skipping")
+            return
+        management.call_command("clearsessions", verbosity=0)
+
+
+@shared_task
+def celery_queue_uptime_check(queue: str) -> None:
+    """No-op heartbeat proving beat dispatches and that `queue` has a consumer."""
+    logger.info("Celery queue '%s' is taking tasks", queue)

deploy/helm/ifrcgo-helm/templates/config/configmap.yaml

@@ -45,7 +45,9 @@ data:
   AUTO_TRANSLATION_TRANSLATOR: {{ .Values.env.AUTO_TRANSLATION_TRANSLATOR | quote }}
   DJANGO_READ_ONLY: {{ .Values.env.DJANGO_READ_ONLY | quote }}
   SENTRY_SAMPLE_RATE: {{ .Values.env.SENTRY_SAMPLE_RATE | quote }}
+  SENTRY_DEBUG: {{ .Values.env.SENTRY_DEBUG | quote }}
   SENTRY_DSN: {{ .Values.env.SENTRY_DSN | quote }}
+  SENTRY_MONITOR_CELERY_BEAT_TASKS: {{ .Values.env.SENTRY_MONITOR_CELERY_BEAT_TASKS | quote }}
   OIDC_ENABLE: {{ .Values.env.OIDC_ENABLE | quote }}
 
   EOAPI_STAC_EXTERNAL_URL: {{ .Values.env.EOAPI_STAC_EXTERNAL_URL | quote }}

deploy/helm/ifrcgo-helm/values.yaml

@@ -46,7 +46,9 @@ env:
   API_FQDN: ''
   FRONTEND_URL: ''
   DEBUG_EMAIL: ''
+  SENTRY_DEBUG: false
   SENTRY_DSN: ''
+  SENTRY_MONITOR_CELERY_BEAT_TASKS: true
   SENTRY_SAMPLE_RATE: ''
   DJANGO_READ_ONLY: ''
   AUTO_TRANSLATION_TRANSLATOR: ''

docker-compose.yml

@@ -143,7 +143,19 @@ services:
   # For development only
   celery:
     <<: *base_server_setup
-    command: python manage.py run_celery_dev
+    restart: unless-stopped
+    command: ./misc/dev/run_worker.sh
+    healthcheck:
+      test: ["CMD-SHELL", "celery -A main inspect ping -d celery@$$HOSTNAME || exit 1"]
+      interval: 30s
+      timeout: 10s
+      retries: 3
+      start_period: 40s
+
+  celery-beat:
+    <<: *base_server_setup
+    restart: unless-stopped
+    command: ./misc/dev/run_worker_beat.sh
 
   # ------------------ Helper CLI Commands
   # Usage: `docker compose run --rm <service-name>`

docs/cronjobs.md

@@ -0,0 +1,97 @@
+# Cronjobs
+
+There are **two** cronjob mechanisms. Use celery beat for new cronjobs.
+
+A job belongs to one mechanism or the other, **never both** —
+`SentryMonitor.validate_config()` asserts that the enum matches `values.yaml`,
+so mixing them breaks it.
+
+## 1. Celery beat — use this for new cronjobs
+
+Schedules are declared in [`main/cronjobs.py`](../main/cronjobs.py) and synced
+into `django_celery_beat` `PeriodicTask` rows when beat starts. `SCHEDULES` is
+the source of truth: remove an entry and its row is deleted on the next start.
+Rows named `manual:*` are left alone, as an escape hatch for one-off tasks
+created through the admin.
+
+Adding one takes two files, with no helm change and no `cron_job_monitor` run:
+
+**1. Write the task in `<app>/tasks.py`**
+
+```python
+@shared_task(soft_time_limit=..., time_limit=...)
+def my_new_job():
+    with redis_lock(RedisLockKey.MY_NEW_JOB) as acquired:
+        if not acquired:
+            return
+        ...
+```
+
+- The lock matters: `CELERY_ACKS_LATE` is on, so a task can be redelivered to
+  another worker if the one running it dies.
+- Time limits go **on the decorator**. `DatabaseScheduler` silently discards
+  `time_limit` / `soft_time_limit` from a schedule entry's options.
+- Don't set `queue` here — it belongs in the schedule entry below.
+
+**2. Add a `CronJob` entry to `SCHEDULES` in `main/cronjobs.py`**
+
+```python
+"my_new_job": CronJob(
+    task="myapp.tasks.my_new_job",
+    schedule=TimeConstants.EVERY_DAY,
+    options=CronJobOption(
+        expire_seconds=TimeConstants.SECONDS_IN_A_DAY,
+        queue=CeleryQueue.cronjob.name,
+    ),
+    sentry_config=CronJobSentryConfig(max_runtime=10),
+),
+```
+
+`options` only supports the keys `ModelEntry._unpack_options` keeps — `queue`,
+`exchange`, `routing_key`, `priority`, `headers`, `expire_seconds`. Anything
+else is dropped without warning. `expire_seconds` stops a backlog accumulating
+while workers are down.
+
+Sentry cron monitoring is automatic, controlled by
+`SENTRY_MONITOR_CELERY_BEAT_TASKS` (default on). `CronJobSentryConfig` sets each
+job's grace period, max runtime and thresholds next to its schedule.
+
+### Queues
+
+`CeleryQueue` in `main/cronjobs.py` declares which queues exist (`default`,
+`heavy`, `cronjob`) and feeds `app.conf.task_queues`. A worker started without
+`-Q` consumes all of them, which is the dev setup.
+
+A queue that is routed to but not declared here is a black hole: the task is
+accepted and then never consumed by anything.
+
+### Running locally
+
+```bash
+docker-compose up celery celery-beat
+```
+
+Worker and beat entrypoints live in `misc/dev/`.
+
+### Not deployed yet
+
+**Beat currently runs in local development only.** There is no beat Deployment
+in `deploy/helm/`, so nothing in `SCHEDULES` fires in alpha/staging/prod until
+one is added. Still to do:
+
+- A beat Deployment with **`replicas: 1`** and `strategy: Recreate` — two beat
+  processes fire every cronjob twice — plus a `celeryBeat` block in
+  `values.yaml`. It needs the same `envFrom` secret + configmap as the celery
+  worker.
+- Beat needs the `django_celery_beat` tables, which `manage.py migrate` creates
+  on the API pod. If beat starts first it crashloops until migrations have run.
+
+## 2. Kubernetes CronJobs — the legacy set
+
+The pre-existing cronjobs run as k8s CronJob resources listed under `cronjobs:`
+in `deploy/helm/ifrcgo-helm/values.yaml`, one pod per run, monitored via
+`SentryMonitor` in `main/sentry.py`. Their Sentry monitors are registered with:
+
+```bash
+docker-compose exec serve bash ./manage.py cron_job_monitor
+```

lang/tasks.py

@@ -12,7 +12,7 @@
 from modeltranslation.translator import translator
 from modeltranslation.utils import build_localized_fieldname
 
-from main.celery import Queues
+from main.cronjobs import CeleryQueue
 from main.lock import RedisLockKey, redis_lock
 from main.translation import (
     TRANSLATOR_ORIGINAL_LANGUAGE_FIELD_NAME,
@@ -217,7 +217,7 @@ def run(self, batch_size=None, only_models: typing.Optional[typing.List[models.M
                 index += 1
 
 
-@shared_task(queue=Queues.CRONJOB)
+@shared_task(queue=CeleryQueue.cronjob.name)
 def translate_remaining_models_fields():
     # Disabled in DEBUG/Development
     if settings.DEBUG:
@@ -226,7 +226,7 @@ def translate_remaining_models_fields():
     ModelTranslator().run(batch_size=100)
 
 
-@shared_task(queue=Queues.DEFAULT)
+@shared_task(queue=CeleryQueue.default.name)
 def translate_model_fields(model_name, pk):
     model = django_apps.get_model(model_name)
     obj = model.objects.get(pk=pk)
@@ -239,7 +239,7 @@ def translate_model_fields(model_name, pk):
         logger.info(f"Translation success for {model_name} with pk={pk}.")
 
 
-@shared_task(queue=Queues.HEAVY)
+@shared_task(queue=CeleryQueue.heavy.name)
 def translate_model_fields_in_bulk(model_name, pks):
     model = django_apps.get_model(model_name)
     qs = model.objects.filter(

main/celery.py

@@ -1,18 +1,16 @@
+import dataclasses
 import os
 
 import celery
 from django.conf import settings
 
-from main import sentry
+from main.cronjobs import BEAT_SCHEDULES, CeleryQueue
 
 
 class CustomCeleryApp(celery.Celery):
     def on_configure(self):
         if settings.SENTRY_DSN:
-            sentry.init_sentry(
-                app_type="WORKER",
-                **settings.SENTRY_CONFIG,
-            )
+            dataclasses.replace(settings.SENTRY_CONFIG, app_type="WORKER").init_sentry()
 
 
 # set the default Django settings module for the 'celery' program.
@@ -30,19 +28,12 @@ def on_configure(self):
 app.autodiscover_tasks()
 
 
-class Queues:
-    DEFAULT = "default"
-    HEAVY = "heavy"
-    CRONJOB = "cronjob"
+app.conf.task_default_queue = CeleryQueue.default.name
+app.conf.task_queues = CeleryQueue.ALL_QUEUE
 
-    DEV_QUEUES = (
-        DEFAULT,
-        HEAVY,
-        CRONJOB,
-    )
-
-
-app.conf.task_default_queue = Queues.DEFAULT
+# Cronjobs scheduled through celery beat. See main/cronjobs.py -- note that this
+# is separate from the legacy k8s CronJob resources in values.yaml:cronjobs.
+app.conf.beat_schedule = BEAT_SCHEDULES
 
 
 @app.task(bind=True)

main/checks.py

@@ -1,7 +1,21 @@
+from pydoc import locate
+
 from django.conf import settings
 from django.core.checks import Error, Tags, register
 
 
+@register(Tags.compatibility)
+def celery_beat_tasks(app_configs, **kwargs):
+    """Catch a typo'd SCHEDULES task path now, not on beat's first tick."""
+    from main.cronjobs import SCHEDULES
+
+    errors = []
+    for name, config in SCHEDULES.items():
+        if locate(config.task) is None:
+            errors.append(Error(f"Celery beat <{name}> task is incorrect: {config.task}"))
+    return errors
+
+
 @register(Tags.compatibility)
 def oauth2_check(app_configs, **kwargs):
     if not settings.OIDC_ENABLE: