feat(celery): add dedicated worker for long-running tasks (#6470)

### 📣 Summary
Introduce a new Celery worker dedicated to handling long-running tasks
separately from regular workloads.

### 📖 Description
This change adds a dedicated Celery worker queue for long-running or
resource-intensive tasks. The goal is to isolate these jobs from
shorter, high-frequency tasks to prevent queue blocking.

Author: noliveleger

docker/entrypoint_celery_kobocat_worker.bash

@@ -14,7 +14,7 @@ exec celery -A kobo worker --loglevel=info \
     --logfile=${KPI_LOGS_DIR}/celery_kobocat_worker.log \
     --pidfile=/tmp/celery_kobocat_worker.pid \
     --queues=kobocat_queue \
-    --exclude-queues=kpi_low_priority_queue,kpi_queue \
+    --exclude-queues=kpi_low_priority_queue,kpi_queue,kpi_long_running_tasks_queue \
     --uid=${UWSGI_USER} \
     --gid=${UWSGI_GROUP} \
     --autoscale ${AUTOSCALE_MIN},${AUTOSCALE_MAX}

docker/entrypoint_celery_kpi_worker.bash

@@ -14,7 +14,7 @@ exec celery -A kobo worker --loglevel=info \
     --logfile=${KPI_LOGS_DIR}/celery_kpi_worker.log \
     --pidfile=/tmp/celery_kpi_worker.pid \
     --queues=kpi_queue \
-    --exclude-queues=kpi_low_priority_queue,kobocat_queue \
+    --exclude-queues=kpi_low_priority_queue,kobocat_queue,kpi_long_running_tasks_queue \
     --uid=${UWSGI_USER} \
     --gid=${UWSGI_GROUP} \
     --autoscale ${AUTOSCALE_MIN},${AUTOSCALE_MAX}

docker/entrypoint_celery_kpi_worker_long_running_tasks.bash

@@ -0,0 +1,20 @@
+#!/bin/bash
+set -e
+source /etc/profile
+
+# Run the Celery worker for long-running jobs ONLY
+
+cd "${KPI_SRC_DIR}"
+
+AUTOSCALE_MIN="${CELERY_AUTOSCALE_MIN:-2}"
+AUTOSCALE_MAX="${CELERY_AUTOSCALE_MAX:-6}"
+
+exec celery -A kobo worker --loglevel=info \
+    --hostname=kpi_worker_long_running_tasks@%h \
+    --logfile=${KPI_LOGS_DIR}/celery_kpi_worker_long_running_tasks.log \
+    --pidfile=/tmp/celery_kpi_worker_long_running_tasks.pid \
+    --queues=kpi_long_running_tasks_queue \
+    --exclude-queues=kpi_queue,kobocat_queue,kpi_low_priority_queue \
+    --uid=${UWSGI_USER} \
+    --gid=${UWSGI_GROUP} \
+    --autoscale ${AUTOSCALE_MIN},${AUTOSCALE_MAX}

docker/entrypoint_celery_kpi_low_priority_worker.bash renamed to docker/entrypoint_celery_kpi_worker_low_priority.bash

@@ -10,11 +10,11 @@ AUTOSCALE_MIN="${CELERY_AUTOSCALE_MIN:-2}"
 AUTOSCALE_MAX="${CELERY_AUTOSCALE_MAX:-6}"
 
 exec celery -A kobo worker --loglevel=info \
-    --hostname=kpi_low_priority_worker@%h \
-    --logfile=${KPI_LOGS_DIR}/celery_kpi_low_priority_worker.log \
-    --pidfile=/tmp/celery_kpi_low_priority_worker.pid \
+    --hostname=kpi_worker_low_priority@%h \
+    --logfile=${KPI_LOGS_DIR}/celery_kpi_worker_low_priority.log \
+    --pidfile=/tmp/celery_kpi_worker_low_priority.pid \
     --queues=kpi_low_priority_queue \
-    --exclude-queues=kpi_queue,kobocat_queue \
+    --exclude-queues=kpi_queue,kobocat_queue,kpi_long_running_tasks_queue \
     --uid=${UWSGI_USER} \
     --gid=${UWSGI_GROUP} \
     --autoscale ${AUTOSCALE_MIN},${AUTOSCALE_MAX}

kobo/apps/long_running_migrations/tasks.py

@@ -9,7 +9,7 @@
 
 
 @celery_app.task(
-    queue='kpi_low_priority_queue',
+    queue='kpi_long_running_tasks_queue',
     soft_time_limit=settings.CELERY_LONG_RUNNING_TASK_SOFT_TIME_LIMIT,
     time_limit=settings.CELERY_LONG_RUNNING_TASK_TIME_LIMIT,
 )

kobo/apps/user_reports/tasks.py

@@ -1,3 +1,5 @@
+import time
+
 from django.conf import settings
 from django.core.cache import cache
 from django.utils import timezone
@@ -17,14 +19,14 @@
     cleanup_stale_snapshots_and_refresh_mv,
     get_or_create_run,
     iter_org_chunks_after,
-    process_chunk,
+    process_chunk, refresh_user_reports_materialized_view,
 )
 from kobo.celery import celery_app
 from kpi.utils.log import logging
 
 
 @celery_app.task(
-    queue='kpi_low_priority_queue',
+    queue='kpi_long_running_tasks_queue',
     soft_time_limit=settings.CELERY_LONG_RUNNING_TASK_SOFT_TIME_LIMIT,
     time_limit=settings.CELERY_LONG_RUNNING_TASK_TIME_LIMIT,
 )
@@ -71,38 +73,66 @@ def refresh_user_report_snapshots(**kwargs):
     if not lock.acquire(blocking=False, blocking_timeout=0):
         logging.info('Nothing to do, task is already running!')
         return
+    else:
+        logging.info('Starting process, refreshing materialized view!')
 
     # Claim the existing snapshot run or create a new one
     run = get_or_create_run()
+
+    # Update last heart-beat
+    BillingAndUsageSnapshotRun.objects.filter(pk=run.pk).update(
+        date_modified=timezone.now()
+    )
+
     last_processed_org_id = run.last_processed_org_id or ''
+    last_time = time.time()
+
     try:
         while chunk_qs := iter_org_chunks_after(last_processed_org_id):
+            logging.info(
+                f'Processing queue, last_processed_org_id: {last_processed_org_id}'
+            )
             billing_map = get_current_billing_period_dates_by_org(chunk_qs)
+            logging.info('\tBilling map retrieved')
             limits_map = get_organizations_effective_limits(chunk_qs, True, True)
+            logging.info('\tLimits map retrieved')
             usage_map = calc.calculate_usage_batch(chunk_qs, billing_map)
+            logging.info('\tUsage map retrieved')
             last_processed_org_id = process_chunk(
                 chunk_qs, usage_map, limits_map, run.pk
             )
 
             # Update the run progress
+            logging.info(
+                f'\tUpdating hearbeat, '
+                f'new last_processed_org_id: {last_processed_org_id}'
+            )
             BillingAndUsageSnapshotRun.objects.filter(pk=run.pk).update(
                 last_processed_org_id=last_processed_org_id,
                 date_modified=timezone.now(),
             )
 
+            if time.time() - last_time >= 15 * 60:
+                logging.info('\tRefreshing the materialized view…')
+                last_time = time.time()
+                refresh_user_reports_materialized_view()
+
         # All orgs processed: cleanup stale, refresh MV and mark run as completed
+        logging.info('Clean-up')
         cleanup_stale_snapshots_and_refresh_mv(run.pk)
+        logging.info('Mark run as complete')
         BillingAndUsageSnapshotRun.objects.filter(pk=run.pk).update(
             status=BillingAndUsageSnapshotStatus.COMPLETED,
             date_modified=timezone.now(),
         )
 
-        # Release the lock
-        lock.release()
-
     except Exception as ex:
         run = BillingAndUsageSnapshotRun.objects.get(pk=run.pk)
         details = run.details or {}
         details.update({'last_error': str(ex), 'ts': timezone.now().isoformat()})
         run.details = details
         run.save(update_fields=['details', 'date_modified'])
+    finally:
+        # Release the lock
+        lock.release()
+        logging.info('Lock released!')

kobo/settings/base.py

@@ -1444,11 +1444,11 @@ def dj_stripe_request_callback_method():
         'schedule': crontab(minute='*/30'),
         'options': {'queue': 'kpi_low_priority_queue'}
     },
-    # Schedule every 30 minutes
+    # Schedule every 15 minutes
     'refresh-user-report-snapshot': {
         'task': 'kobo.apps.user_reports.tasks.refresh_user_report_snapshots',
-        'schedule': crontab(minute='*/30'),
-        'options': {'queue': 'kpi_low_priority_queue'},
+        'schedule': crontab(minute='*/15'),
+        'options': {'queue': 'kpi_long_running_tasks_queue'},
     },
     # Schedule every day at midnight UTC
     'project-ownership-garbage-collector': {