diff --git a/README.md b/README.md index 51f07f0e8..4dde43a0d 100644 --- a/README.md +++ b/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 ``` diff --git a/api/management/commands/run_celery_dev.py b/api/management/commands/run_celery_dev.py index a65a04a51..a0300d4f0 100644 --- a/api/management/commands/run_celery_dev.py +++ b/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(): diff --git a/api/management/commands/run_celery_prod.py b/api/management/commands/run_celery_prod.py index 7278ec322..28dded685 100644 --- a/api/management/commands/run_celery_prod.py +++ b/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. diff --git a/api/tasks.py b/api/tasks.py index 411473b64..bad694ef8 100644 --- a/api/tasks.py +++ b/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) diff --git a/deploy/helm/ifrcgo-helm/templates/config/configmap.yaml b/deploy/helm/ifrcgo-helm/templates/config/configmap.yaml index e283b36cd..ac95b4d68 100644 --- a/deploy/helm/ifrcgo-helm/templates/config/configmap.yaml +++ b/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 }} diff --git a/deploy/helm/ifrcgo-helm/values.yaml b/deploy/helm/ifrcgo-helm/values.yaml index f4f3f072f..f6543aeb3 100644 --- a/deploy/helm/ifrcgo-helm/values.yaml +++ b/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: '' diff --git a/docker-compose.yml b/docker-compose.yml index 2aa673ec4..e5b79f01c 100644 --- a/docker-compose.yml +++ b/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 ` diff --git a/docs/cronjobs.md b/docs/cronjobs.md new file mode 100644 index 000000000..b604869fb --- /dev/null +++ b/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 `/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 +``` diff --git a/lang/tasks.py b/lang/tasks.py index e27947ffd..bf0481659 100644 --- a/lang/tasks.py +++ b/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( diff --git a/main/celery.py b/main/celery.py index a6ac4f13b..113d6d87d 100644 --- a/main/celery.py +++ b/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) diff --git a/main/checks.py b/main/checks.py index 39c2bf38a..e21c8cfb7 100644 --- a/main/checks.py +++ b/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: diff --git a/main/cronjobs.py b/main/cronjobs.py new file mode 100644 index 000000000..334a12d92 --- /dev/null +++ b/main/cronjobs.py @@ -0,0 +1,199 @@ +""" +Registry for cronjobs scheduled by celery beat. For NEW cronjobs only. + +The legacy cronjobs are k8s CronJobs listed in values.yaml:cronjobs and +monitored via main.sentry.SentryMonitor. A job belongs to one mechanism or the +other, never both -- SentryMonitor.validate_config() asserts the enum matches +values.yaml. See docs/cronjobs.md for how to add one. +""" + +import functools +import logging +import operator +import typing + +from celery import signals +from celery.schedules import crontab +from django.conf import settings +from django.db import models +from kombu import Queue +from sentry_sdk.integrations.celery import beat as sentry_celery_beat + +if typing.TYPE_CHECKING: + from celery import Celery + from sentry_sdk._types import MonitorConfig + +logger = logging.getLogger(__name__) + + +class TimeConstants: + SECONDS_IN_A_MINUTE = 60 + SECONDS_IN_A_HOUR = 60 * 60 + SECONDS_IN_A_DAY = 24 * 60 * 60 + SECONDS_IN_A_WEEK = 7 * 24 * 60 * 60 + + EVERY_WEEK = crontab(minute="1", hour="1", day_of_week="1") + EVERY_DAY = crontab(minute="1", hour="1") + EVERY_HOUR = crontab(minute="0", hour="*") + EVERY_2_MINUTES = crontab(minute="*/2") + EVERY_1_MINUTES = crontab(minute="*/1") + + +class CeleryQueue: + # NOTE: Names must be lowercase (used as-is in k8s). + default = Queue("default") + heavy = Queue("heavy") + cronjob = Queue("cronjob") + + # Feeds app.conf.task_queues. A worker started without -Q consumes all of these. + ALL_QUEUE = ( + default, + heavy, + cronjob, + ) + + +class CronJobOption(typing.TypedDict, total=False): + """Per-job options for django_celery_beat. + + WARNING: only queue/exchange/routing_key/priority/headers/expire_seconds + survive ModelEntry._unpack_options; anything else is silently dropped. Put + time_limit/soft_time_limit on the task decorator instead. + """ + + expire_seconds: float + """Task will not run if picked up later than this. Avoids a backlog when workers are down.""" + + queue: str + """Queue the task is sent to. Required -- see test_cronjobs.py.""" + + +class CeleryBeatSchedule(typing.TypedDict): + task: str + schedule: crontab + options: CronJobOption + args: tuple[typing.Any, ...] + + +class CronJobSentryConfig(typing.NamedTuple): + checkin_margin: int = 5 + """Minutes of grace before a missed check-in is flagged.""" + + max_runtime: int = 30 + """Minutes before Sentry considers the job failed.""" + + failure_issue_threshold: int = 1 + """Consecutive failures before an issue is created.""" + + recovery_threshold: int = 1 + """Consecutive successes before an issue is resolved.""" + + def as_dict(self) -> "MonitorConfig": + return { + "checkin_margin": self.checkin_margin, + "max_runtime": self.max_runtime, + "failure_issue_threshold": self.failure_issue_threshold, + "recovery_threshold": self.recovery_threshold, + } + + +class CronJob(typing.NamedTuple): + task: str + schedule: crontab + args: tuple[typing.Any, ...] = () + sentry_config: CronJobSentryConfig = CronJobSentryConfig() + options: CronJobOption = {} + + +# NOTE: Removing an entry here deletes its PeriodicTask row (see update_periodic_tasks). +SCHEDULES: dict[str, CronJob] = { + "clear_expired_django_sessions": CronJob( + task="api.tasks.clear_expired_django_sessions", + schedule=TimeConstants.EVERY_WEEK, + options=CronJobOption(expire_seconds=TimeConstants.SECONDS_IN_A_WEEK), + ), + **{ + f"celery_queue_uptime_{celery_queue.name}": CronJob( + task="api.tasks.celery_queue_uptime_check", + args=(celery_queue.name,), + schedule=TimeConstants.EVERY_HOUR, + options=CronJobOption( + expire_seconds=TimeConstants.SECONDS_IN_A_HOUR, + queue=celery_queue.name, + ), + sentry_config=CronJobSentryConfig( + checkin_margin=10, + max_runtime=2, + failure_issue_threshold=2, + ), + ) + for celery_queue in CeleryQueue.ALL_QUEUE + }, +} + +BEAT_SCHEDULES: dict[str, CeleryBeatSchedule] = { + name: { + "task": config.task, + "args": config.args, + "schedule": config.schedule, + "options": config.options, + } + for name, config in SCHEDULES.items() +} + + +_get_monitor_config = sentry_celery_beat._get_monitor_config + + +class SentryMonkeyPatch: + @staticmethod + def custom__get_monitor_config(celery_schedule: typing.Any, app: "Celery", monitor_name: str) -> "MonitorConfig": + """Get configuration for sentry monitoring. + + https://github.com/getsentry/sentry-python/blob/5715734eac1c5fb4b6ec61ef459080c74fa777b5/sentry_sdk/integrations/celery/beat.py#L59 + """ + config = _get_monitor_config(celery_schedule, app, monitor_name) + job_config = SCHEDULES.get(monitor_name) + if job_config: + config.update(job_config.sentry_config.as_dict()) + return config + + +sentry_celery_beat._get_monitor_config = SentryMonkeyPatch.custom__get_monitor_config + + +def get_obsolete_periodic_task_qs(): + """Our PeriodicTask rows that are no longer in SCHEDULES. + + Scoped to GO_APPS so third-party rows are never touched. `manual:` rows are + spared as an escape hatch for one-off tasks created in the admin. + """ + from django_celery_beat.models import PeriodicTask + + ours = functools.reduce( + operator.or_, + [models.Q(task__startswith=f"{app_name}.") for app_name in settings.GO_APPS], + ) + + return PeriodicTask.objects.filter(ours).exclude(name__in=list(BEAT_SCHEDULES.keys())).exclude(name__startswith="manual:") + + +@signals.beat_init.connect +def update_periodic_tasks(**_): + """Drop rows for cronjobs no longer in SCHEDULES -- git is the source of truth.""" + logger.info("Cronjob sync: Start") + try: + obsolete_tasks_qs = get_obsolete_periodic_task_qs() + + obsolete_task_names = list(obsolete_tasks_qs.values_list("name", flat=True)) + if not obsolete_task_names: + logger.info("Cronjob sync - Obsolete tasks: Nothing to do") + return + + for task_name in obsolete_task_names: + logger.warning("Cronjob sync - Obsolete tasks: Task <%s> will be deleted", task_name) + + deleted_count, _ = obsolete_tasks_qs.delete() + logger.warning("Cronjob sync - Obsolete tasks: Deleted %s tasks", deleted_count) + except Exception: + logger.error("Cronjob sync: Failed to sync PeriodicTasks", exc_info=True) diff --git a/main/lock.py b/main/lock.py index 3d1fd2de1..11b2244a9 100644 --- a/main/lock.py +++ b/main/lock.py @@ -20,22 +20,27 @@ class RedisLockKey(models.TextChoices): OPERATION_LEARNING_SUMMARY = _BASE + "-operation-learning-summary-{0}" OPERATION_LEARNING_SUMMARY_EXPORT = _BASE + "-operation-learning-summary-export-{0}" MODEL_TRANSLATION = _BASE + "-{model_name}-translation-{id}" + CLEAR_EXPIRED_DJANGO_SESSIONS = _BASE + "-clear-expired-django-sessions" @contextmanager def redis_lock( key: RedisLockKey, - id: typing.Union[int, str], + id: typing.Union[int, str, None] = None, model_name: typing.Optional[str] = None, lock_expire: int = settings.REDIS_DEFAULT_LOCK_EXPIRE, ): """ Locking mechanism using Redis + + `id` is optional, for singleton locks whose key has no placeholder. """ if model_name and id: lock_id = key.format(model_name=model_name, id=id) - else: + elif id is not None: lock_id = key.format(id) + else: + lock_id = str(key) timeout_at = time.monotonic() + lock_expire - 3 # cache.add fails if the key already exists status = cache.add(lock_id, 1, lock_expire) diff --git a/main/sentry.py b/main/sentry.py index d21eb02fd..40f4ed91d 100644 --- a/main/sentry.py +++ b/main/sentry.py @@ -1,3 +1,4 @@ +import dataclasses import logging import os import typing @@ -82,21 +83,48 @@ def fetch_git_sha(path, head=None): return str(fh.read()).strip() -def init_sentry(app_type, tags={}, **config): - integrations = [ - CeleryIntegration(), - DjangoIntegration(), - RedisIntegration(), - ] - sentry_sdk.init( - **config, - ignore_errors=IGNORED_ERRORS, - integrations=integrations, - ) - with sentry_sdk.configure_scope() as scope: - scope.set_tag("app_type", app_type) - for tag, value in tags.items(): - scope.set_tag(tag, value) +@dataclasses.dataclass +class SentryConfig: + """Typed sentry options. + + Every option is passed to sentry_sdk.init() by name, so a config key it does + not recognise can no longer reach it. Use dataclasses.replace() to vary a + field per process (see main/celery.py). + """ + + dsn: str + release: str | None + environment: str + send_default_pii: bool + traces_sample_rate: float + enable_tracing: bool + debug: bool + # Custom configs + monitor_celery_beat_tasks: bool + app_type: str + tags: dict[str, str] + + def init_sentry(self): + integrations = [ + DjangoIntegration(), + RedisIntegration(), + CeleryIntegration(monitor_beat_tasks=self.monitor_celery_beat_tasks), + ] + sentry_sdk.init( + ignore_errors=IGNORED_ERRORS, + integrations=integrations, + dsn=self.dsn, + release=self.release, + environment=self.environment, + send_default_pii=self.send_default_pii, + traces_sample_rate=self.traces_sample_rate, + enable_tracing=self.enable_tracing, + debug=self.debug, + ) + with sentry_sdk.configure_scope() as scope: + scope.set_tag("app_type", self.app_type) + for tag, value in self.tags.items(): + scope.set_tag(tag, value) class SentryMonitor(models.TextChoices): diff --git a/main/settings.py b/main/settings.py index 926ff3d9a..f7598e629 100644 --- a/main/settings.py +++ b/main/settings.py @@ -1,3 +1,4 @@ +# type: ignore[reportAttributeAccessIssue] import base64 import logging import os @@ -117,9 +118,12 @@ APPEALS_USER=(str, None), APPEALS_PASS=(str, None), # Sentry + SENTRY_DEBUG=(bool, False), SENTRY_DSN=(str, None), SENTRY_SAMPLE_RATE=(float, 0.2), SENTRY_RELEASE=(str, None), + # Sentry cron monitors for celery beat tasks only; legacy k8s CronJobs use SentryMonitor. + SENTRY_MONITOR_CELERY_BEAT_TASKS=(bool, True), # Maintenance mode DJANGO_READ_ONLY=(bool, False), # Misc @@ -276,6 +280,7 @@ def parse_domain(*env_keys: str) -> str: # GO Apps *GO_APPS, # Utils Apps + "django_celery_beat", "oauth2_provider", "tinymce", "admin_auto_filters", @@ -759,6 +764,8 @@ def log_render_extra_context(record): CELERY_RESULT_BACKEND = CELERY_REDIS_URL CELERY_TIMEZONE = TIME_ZONE CELERY_ACKS_LATE = True +# Schedules come from main/cronjobs.py, synced into PeriodicTask rows on beat startup. +CELERY_BEAT_SCHEDULER = "django_celery_beat.schedulers:DatabaseScheduler" RETRY_STRATEGY = Retry(total=3, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["HEAD", "GET", "OPTIONS"]) @@ -806,9 +813,12 @@ def log_render_extra_context(record): LAST_GIT_TAG = 0 # Sentry Config +SENTRY_DEBUG = env("SENTRY_DEBUG") SENTRY_DSN = env("SENTRY_DSN") SENTRY_SAMPLE_RATE = env("SENTRY_SAMPLE_RATE") +SENTRY_MONITOR_CELERY_BEAT_TASKS = env("SENTRY_MONITOR_CELERY_BEAT_TASKS") + SENTRY_RELEASE = env("SENTRY_RELEASE") if not SENTRY_RELEASE: try: @@ -817,23 +827,26 @@ def log_render_extra_context(record): # .git is missing, unreadable, or a gitfile pointer (submodule/worktree checkout) SENTRY_RELEASE = "unknown" -SENTRY_CONFIG = { - "dsn": SENTRY_DSN, - "send_default_pii": True, - "traces_sample_rate": SENTRY_SAMPLE_RATE, - "enable_tracing": True, - "release": SENTRY_RELEASE, - "environment": GO_ENVIRONMENT, - "debug": DEBUG, - "tags": { +SENTRY_CONFIG = sentry.SentryConfig( + dsn=SENTRY_DSN, + release=SENTRY_RELEASE, + environment=GO_ENVIRONMENT, + send_default_pii=True, + traces_sample_rate=SENTRY_SAMPLE_RATE, + enable_tracing=True, + debug=SENTRY_DEBUG, + # Custom configs + monitor_celery_beat_tasks=SENTRY_MONITOR_CELERY_BEAT_TASKS, + # TODO: changes it to env! + app_type="API", + tags={ "site": GO_API_URL, }, -} +) + if SENTRY_DSN: - sentry.init_sentry( - app_type="API", - **SENTRY_CONFIG, - ) + SENTRY_CONFIG.init_sentry() + # Required for Django HayStack HAYSTACK_CONNECTIONS = { "default": { diff --git a/main/test_cronjobs.py b/main/test_cronjobs.py new file mode 100644 index 000000000..0d1ea52ae --- /dev/null +++ b/main/test_cronjobs.py @@ -0,0 +1,123 @@ +import dataclasses +import inspect +from pydoc import locate + +import pytest +from django.conf import settings + +from main import cronjobs +from main.cronjobs import ( + BEAT_SCHEDULES, + SCHEDULES, + CeleryQueue, + SentryMonkeyPatch, + get_obsolete_periodic_task_qs, +) + +ALL_QUEUE_NAMES = {queue.name for queue in CeleryQueue.ALL_QUEUE} + +# SentryConfig fields that are ours, not sentry_sdk.init() options. +CUSTOM_SENTRY_FIELDS = {"monitor_celery_beat_tasks", "app_type", "tags"} + +# The only keys ModelEntry._unpack_options keeps; anything else is silently dropped. +SUPPORTED_OPTION_KEYS = { + "queue", + "exchange", + "routing_key", + "priority", + "headers", + "expire_seconds", +} + + +def test_sentry_config_fields_are_all_accounted_for(): + """Each SentryConfig field must be a real sentry_sdk option or a known custom one. + + sentry_sdk.init() raises `TypeError: Unknown option` for anything it does not + recognise, and that only happens once SENTRY_DSN is set -- never locally. + A field that is neither must be handled by hand in init_sentry(). + """ + from sentry_sdk.consts import DEFAULT_OPTIONS + + from main.sentry import SentryConfig + + fields = {f.name for f in dataclasses.fields(SentryConfig)} + unaccounted = fields - set(DEFAULT_OPTIONS) - CUSTOM_SENTRY_FIELDS + assert not unaccounted, f"SentryConfig fields sentry_sdk.init() does not know: {unaccounted}" + + +def test_sentry_config_still_carries_beat_monitoring(): + assert isinstance(settings.SENTRY_CONFIG.monitor_celery_beat_tasks, bool) + assert dataclasses.replace(settings.SENTRY_CONFIG, app_type="WORKER").app_type == "WORKER" + + +def test_schedules_tasks_are_importable_celery_tasks(): + for name, config in SCHEDULES.items(): + task = locate(config.task) + assert task is not None, f"{name}: task path does not exist: {config.task}" + assert hasattr(task, "delay"), f"{name}: {config.task} is not a celery task" + + +def test_schedules_set_the_queue_in_config_not_on_the_task(): + """A queue is optional -- without one the task uses task_default_queue. When + it is set it belongs in the SCHEDULES entry, never on the task decorator. + """ + for name, config in SCHEDULES.items(): + task_queue = getattr(locate(config.task), "queue", None) + assert task_queue is None, f"{name}: set queue in SCHEDULES, not on the {config.task} decorator" + + queue = config.options.get("queue") + if queue is not None: + assert queue in ALL_QUEUE_NAMES, f"{name}: queue {queue!r} is not one of {ALL_QUEUE_NAMES}" + + +def test_schedules_declare_an_expiry(): + """Without expire_seconds a backlog accumulates while workers are down.""" + missing = [name for name, config in SCHEDULES.items() if "expire_seconds" not in config.options] + assert not missing, f"Cronjobs missing expire_seconds: {missing}" + + +def test_schedules_only_use_options_the_scheduler_honours(): + """Guard against silently-dropped options (eg. time_limit belongs on the task).""" + for name, entry in BEAT_SCHEDULES.items(): + unsupported = set(entry["options"]) - SUPPORTED_OPTION_KEYS + assert not unsupported, f"{name}: options {unsupported} are ignored by DatabaseScheduler" + + +def test_prune_filter_matches_how_our_tasks_are_named(): + """A task path not starting with a GO_APPS label is invisible to pruning.""" + for name, config in SCHEDULES.items(): + app_label = config.task.split(".")[0] + assert app_label in settings.GO_APPS, f"{name}: {config.task} does not start with a GO_APPS label" + + +def test_sentry_monkeypatch_is_applied_and_signature_still_matches(): + """Fails if a sentry-sdk upgrade moves or re-signatures the private hook.""" + from sentry_sdk.integrations.celery import beat as sentry_celery_beat + + assert sentry_celery_beat._get_monitor_config is SentryMonkeyPatch.custom__get_monitor_config + + expected = ["celery_schedule", "app", "monitor_name"] + assert list(inspect.signature(cronjobs._get_monitor_config).parameters) == expected + assert list(inspect.signature(SentryMonkeyPatch.custom__get_monitor_config).parameters) == expected + + +@pytest.mark.django_db +def test_prune_targets_only_obsolete_rows_of_ours(): + from django_celery_beat.models import IntervalSchedule, PeriodicTask + + schedule, _ = IntervalSchedule.objects.get_or_create(every=1, period=IntervalSchedule.DAYS) + + PeriodicTask.objects.create(name="obsolete_job", task="api.tasks.removed_job", interval=schedule) + # Spared: deliberate admin-created escape hatch. + PeriodicTask.objects.create(name="manual:adhoc_job", task="api.tasks.removed_job", interval=schedule) + # Spared: not one of our apps. + PeriodicTask.objects.create(name="vendor_job", task="some_vendor.tasks.thing", interval=schedule) + # Spared: still declared in SCHEDULES. + PeriodicTask.objects.create( + name="clear_expired_django_sessions", + task="api.tasks.clear_expired_django_sessions", + interval=schedule, + ) + + assert set(get_obsolete_periodic_task_qs().values_list("name", flat=True)) == {"obsolete_job"} diff --git a/misc/dev/run_worker.sh b/misc/dev/run_worker.sh new file mode 100755 index 000000000..0553b971c --- /dev/null +++ b/misc/dev/run_worker.sh @@ -0,0 +1,7 @@ +#!/bin/bash -e +# Uses the management command for Django's autoreloader. + +# TODO(susilnem): Use wait_for_resources once banjo-stack is implemented +./misc/wait-for-broker.sh + +exec ./manage.py run_celery_dev diff --git a/misc/dev/run_worker_beat.sh b/misc/dev/run_worker_beat.sh new file mode 100755 index 000000000..addd48af6 --- /dev/null +++ b/misc/dev/run_worker_beat.sh @@ -0,0 +1,5 @@ +#!/bin/bash -e + +./misc/wait-for-broker.sh + +exec celery -A main beat -l INFO diff --git a/misc/wait-for-broker.sh b/misc/wait-for-broker.sh new file mode 100755 index 000000000..5c1f0cea1 --- /dev/null +++ b/misc/wait-for-broker.sh @@ -0,0 +1,16 @@ +#!/bin/bash -e +# Wait for the celery broker. main/entrypoint.sh waits for the db, not the broker. +# +# TODO: Replace with wait_for_resources once banjo-stack is implemented + +BROKER_HOST_PORT=$(python -c " +import os +from urllib.parse import urlparse + +url = urlparse(os.environ['CELERY_REDIS_URL']) +print(f'{url.hostname}:{url.port or 6379}') +") + +echo "Waiting for celery broker at ${BROKER_HOST_PORT}..." +wait-for-it "${BROKER_HOST_PORT}" +>&2 echo "Celery broker is up - continuing..." diff --git a/pyproject.toml b/pyproject.toml index 43462d443..c2778dfb0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -69,7 +69,9 @@ dependencies = [ "xmltodict==0.11.0", "xhtml2pdf==0.2.17", "reportlab", # XXX: Used by xhtml2pdf reportlab==3.6.7 breaks for now - "celery[redis]>=5.2.2", + # NOTE: floor is >=5.2.3 because django-celery-beat requires celery>=5.2.3,<6.0 + "celery[redis]>=5.2.3", + "django-celery-beat>=2.9.0", "django-redis==5.0.0", "sentry-sdk", "django-haystack[elasticsearch]", diff --git a/uv.lock b/uv.lock index 4b73278c2..b906880fa 100644 --- a/uv.lock +++ b/uv.lock @@ -401,6 +401,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b4/d9/e70c286c979378f061d8266e279b686ab0b0b688e1fe0af864684f23a77d/coverage-7.15.4-py3-none-any.whl", hash = "sha256:964730a1e9de9c0cf11be6a1a3c79ce419c34882842abd256086ba4698705e84", size = 214332, upload-time = "2026-08-06T13:50:22.192Z" }, ] +[[package]] +name = "cron-descriptor" +version = "1.4.5" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/02/83/70bd410dc6965e33a5460b7da84cf0c5a7330a68d6d5d4c3dfdb72ca117e/cron_descriptor-1.4.5.tar.gz", hash = "sha256:f51ce4ffc1d1f2816939add8524f206c376a42c87a5fca3091ce26725b3b1bca", size = 30666, upload-time = "2024-08-24T18:16:48.654Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/88/20/2cfe598ead23a715a00beb716477cfddd3e5948cf203c372d02221e5b0c6/cron_descriptor-1.4.5-py3-none-any.whl", hash = "sha256:736b3ae9d1a99bc3dbfc5b55b5e6e7c12031e7ba5de716625772f8b02dcd6013", size = 50370, upload-time = "2024-08-24T18:16:46.783Z" }, +] + [[package]] name = "cryptography" version = "50.0.0" @@ -510,6 +519,23 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e9/18/35bd4e459bfd387c67f1439fafb7e923fa1df620d41a5eae148fa9f5b551/django_admin_list_filter_dropdown-1.0.3-py3-none-any.whl", hash = "sha256:bf1b48bab9772dad79db71efef17e78782d4f2421444d5e49bb10e0da71cd6bb", size = 3813, upload-time = "2019-10-14T10:21:25.963Z" }, ] +[[package]] +name = "django-celery-beat" +version = "2.9.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "celery" }, + { name = "cron-descriptor" }, + { name = "django" }, + { name = "django-timezone-field" }, + { name = "python-crontab" }, + { name = "tzdata" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/05/45/fc97bc1d9af8e7dc07f1e37044d9551a30e6793249864cef802341e2e3a8/django_celery_beat-2.9.0.tar.gz", hash = "sha256:92404650f52fcb44cf08e2b09635cb1558327c54b1a5d570f0e2d3a22130934c", size = 177667, upload-time = "2026-02-28T16:45:34.749Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/71/ae/9befa7ae37f5e5c41be636a254fcf47ff30dd5c88bd115070e252f6b9162/django_celery_beat-2.9.0-py3-none-any.whl", hash = "sha256:4a9e5ebe26d6f8d7215e1fc5c46e466016279dc102435a28141108649bdf2157", size = 105013, upload-time = "2026-02-28T16:45:32.822Z" }, +] + [[package]] name = "django-cors-headers" version = "3.11.0" @@ -763,6 +789,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ed/41/10b8d576651af4f1dc6bdfb0d8a203e8b3a55d60ae36ba305d0edcdb594f/django_stubs_ext-6.1.0-py3-none-any.whl", hash = "sha256:57273506823274700a707c8f404dfb52a12c4554daf1468dfb2bff857bb22074", size = 10404, upload-time = "2026-08-12T10:54:42.979Z" }, ] +[[package]] +name = "django-timezone-field" +version = "7.2.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "django" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/12/0b/22654abc2355f3b84e3e5b9d26569639c15d04b96d7186f3a477ec45c1ef/django_timezone_field-7.2.2.tar.gz", hash = "sha256:a004d0b19fe10bf5964cb21a65b36324b16a61879e4711c0dafdf8d6253e8ebc", size = 13158, upload-time = "2026-06-06T05:28:23.638Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/2b/5f/c8dfb67105c4ef391a3e9d0bcd35b7eae4c4c9d023d612c3999ae1cb32ef/django_timezone_field-7.2.2-py3-none-any.whl", hash = "sha256:30354d0f37462a0b9b5c289e271580a6be9b58dea30e7bf88435372882c8fa7a", size = 13322, upload-time = "2026-06-06T05:28:22.454Z" }, +] + [[package]] name = "django-tinymce" version = "4.1.0" @@ -935,6 +973,7 @@ dependencies = [ { name = "django" }, { name = "django-admin-autocomplete-filter" }, { name = "django-admin-list-filter-dropdown" }, + { name = "django-celery-beat" }, { name = "django-cors-headers" }, { name = "django-coverage" }, { name = "django-enumfield" }, @@ -1020,7 +1059,7 @@ requires-dist = [ { name = "azure-identity" }, { name = "beautifulsoup4", specifier = ">=4.12.3,<5.0" }, { name = "boto3", specifier = ">=1.34.0,<2.0.0" }, - { name = "celery", extras = ["redis"], specifier = ">=5.2.2" }, + { name = "celery", extras = ["redis"], specifier = ">=5.2.3" }, { name = "choicesenum", specifier = ">=0.7.0,<1.0" }, { name = "colorlog" }, { name = "coverage", specifier = ">=7.4.0,<8.0" }, @@ -1028,6 +1067,7 @@ requires-dist = [ { name = "django", specifier = ">=5.2,<5.3" }, { name = "django-admin-autocomplete-filter" }, { name = "django-admin-list-filter-dropdown" }, + { name = "django-celery-beat", specifier = ">=2.9.0" }, { name = "django-cors-headers", specifier = "==3.11.0" }, { name = "django-coverage", specifier = "==1.2.4" }, { name = "django-enumfield", specifier = "==2.0.2" }, @@ -2238,6 +2278,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c7/f0/f6b9d17e3426e7b54c18d05d917982647a976233ed9348a2ef40f1a17f85/python_bidi-0.6.11-pp311-pypy311_pp73-musllinux_1_2_x86_64.whl", hash = "sha256:cf4e88a6fec81b7155a487cbbea7753a3d9a76dc4d391b4f8958b37227ef2c12", size = 505204, upload-time = "2026-06-30T14:23:41.222Z" }, ] +[[package]] +name = "python-crontab" +version = "3.3.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/99/7f/c54fb7e70b59844526aa4ae321e927a167678660ab51dda979955eafb89a/python_crontab-3.3.0.tar.gz", hash = "sha256:007c8aee68dddf3e04ec4dce0fac124b93bd68be7470fc95d2a9617a15de291b", size = 57626, upload-time = "2025-07-13T20:05:35.535Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/47/42/bb4afa5b088f64092036221843fc989b7db9d9d302494c1f8b024ee78a46/python_crontab-3.3.0-py3-none-any.whl", hash = "sha256:739a778b1a771379b75654e53fd4df58e5c63a9279a63b5dfe44c0fcc3ee7884", size = 27533, upload-time = "2025-07-13T20:05:34.266Z" }, +] + [[package]] name = "python-dateutil" version = "2.9.0.post0"