UN-3974 [DEV] Cut dashboard cron DB time by splitting the metrics schedule by tier and indexing workflow_execution on created_at

backend/dashboard_metrics/README.md

@@ -14,7 +14,7 @@ This module provides a metrics dashboard for monitoring document processing, API
 ### Data Flow
 ```
 Source Tables (usage_v2, page_usage, workflow_execution, workflow_file_execution)
-       ↓ [Celery task every 15 min]
+       ↓ [Celery: hourly tier every 15 min, daily+monthly hourly at :20]
 Aggregated Tables (EventMetricsHourly → Daily → Monthly)
        ↓
 API Endpoints (/overview/, /summary/, /series/)
@@ -46,8 +46,9 @@ celery -A backend beat -l info
 ### Celery Tasks & Schedule
 | Task | Schedule | What It Does |
 |------|----------|--------------|
-| `aggregate_from_sources` | Every 15 min | Aggregates source → hourly/daily; rolls monthly up from daily |
-| `aggregate_from_sources` (reconcile) | Daily 4:40 AM | Same task over a 7-day source window, to repair gaps after downtime |
+| `aggregate_from_sources` | Every 15 min | Aggregates source → **hourly tier only** (`tier=hourly`) |
+| `aggregate_daily_monthly` | Hourly at :20 | Aggregates source → daily; rolls monthly up from daily (`tier=daily_monthly`) |
+| `aggregate_from_sources` (reconcile) | Daily 4:40 AM | All tiers over a 7-day source window, to repair gaps after downtime |
 | `cleanup_hourly_data` | Daily 2 AM | Deletes hourly data > 30 days |
 | `cleanup_daily_data` | Weekly Sun 3 AM | Deletes daily data > 365 days |
 
@@ -163,7 +164,7 @@ The dashboard reads from **pre-aggregated tables** (`event_metrics_hourly`, `eve
 - Source table performance is unaffected by the dashboard feature. If the aggregation task is slow or fails, source tables continue working normally.
 
 **Failure Resilience:**
-- If the aggregation task fails, the dashboard shows stale data (up to 15 minutes old) rather than crashing.
+- If the aggregation task fails, the dashboard shows stale data rather than crashing — up to 15 minutes old for hourly figures, up to an hour for daily and monthly.
 - A daily 04:40 UTC reconciliation pass reruns the same task over a 7-day source window, so a gap shorter than that repairs itself without a manual backfill.
 - Celery tasks have `max_retries=3` with exponential backoff.
 - Cleanup tasks (hourly: 30-day retention, daily: 365-day retention) prevent unbounded table growth.
@@ -341,8 +342,9 @@ Located in `tasks.py`:
 
 | Task Name | Celery Name | Schedule | Queue | Purpose |
 |-----------|-------------|----------|-------|---------|
-| `aggregate_metrics_from_sources` | `dashboard_metrics.aggregate_from_sources` | Every 15 min | `dashboard_metric_events` | Aggregate from source tables |
-| `aggregate_metrics_from_sources` | `dashboard_metrics.aggregate_from_sources` | Daily 4:40 AM UTC | `dashboard_metric_events` | Reconciliation pass, `source_window_days=7` |
+| `aggregate_metrics_from_sources` | `dashboard_metrics.aggregate_from_sources` | Every 15 min | `dashboard_metric_events` | Aggregate the hourly tier (`tier=hourly`) |
+| `aggregate_metrics_from_sources` | `dashboard_metrics.aggregate_from_sources` | Hourly at :20 UTC | `dashboard_metric_events` | Aggregate the daily and monthly tiers (`tier=daily_monthly`) |
+| `aggregate_metrics_from_sources` | `dashboard_metrics.aggregate_from_sources` | Daily 4:40 AM UTC | `dashboard_metric_events` | Reconciliation pass, all tiers, `source_window_days=7` |
 | `cleanup_hourly_metrics` | `dashboard_metrics.cleanup_hourly_data` | Daily 2:00 AM UTC | `dashboard_metric_events` | Delete hourly data >30 days |
 | `cleanup_daily_metrics` | `dashboard_metrics.cleanup_daily_data` | Weekly Sun 3:00 AM UTC | `dashboard_metric_events` | Delete daily data >365 days |
 

backend/dashboard_metrics/internal_views.py

@@ -35,6 +35,7 @@
 
 from dashboard_metrics.tasks import (
     DASHBOARD_SOURCE_WINDOW_DAYS,
+    AggregationTier,
     aggregate_metrics_from_sources,
     cleanup_daily_metrics,
     cleanup_hourly_metrics,
@@ -59,6 +60,15 @@ def _clear_org_context() -> None:
         StateStore.clear(Account.ORGANIZATION_ID)
 
 
+def _tier_arg(raw: Any) -> AggregationTier:
+    """Coerce a request body's tier. An explicit null means "omitted", not "none"."""
+    try:
+        return AggregationTier(raw)
+    except ValueError as exc:
+        valid = [member.value for member in AggregationTier]
+        raise ValueError(f"tier must be one of {valid}, got {raw!r}") from exc
+
+
 def _int_arg(request: Request, key: str, default: int) -> int:
     """Read an optional positive integer from the request body."""
     raw = request.data.get(key, default) if isinstance(request.data, dict) else default
@@ -75,11 +85,12 @@ class _MetricsTaskAPIView(APIView):
     """Shared plumbing: clear org context, run, translate errors."""
 
     def _run(self, fn, *args: Any, **kwargs: Any) -> Response:
+        """Run one task body. Every view validates its own body first, so anything
+        raising in here is an internal fault and belongs on the logged 500 path.
+        """
         _clear_org_context()
         try:
             return Response(fn(*args, **kwargs))
-        except ValueError as exc:  # bad request body
-            return Response({"error": str(exc)}, status=status.HTTP_400_BAD_REQUEST)
         except Exception as exc:
             logger.error("dashboard-metrics internal call failed: %s", exc, exc_info=True)
             return Response(
@@ -92,20 +103,29 @@ class AggregateMetricsAPIView(_MetricsTaskAPIView):
 
     Calls the Celery task body verbatim, Redis lock included — this endpoint exists
     only because the PG consumer has no Django, not to change what the job does.
+
+    Two optional body fields, both validated here at the boundary so a ValueError
+    from inside the ten-minute aggregation stays a logged 500 rather than reading as
+    a bad request: ``tier`` selects which tiers to write, ``source_window_days``
+    widens the daily lookback for the reconciliation pass. Omitting either — or
+    sending it as ``null`` — applies the task's own default.
     """
 
     def post(self, request: Request) -> Response:
-        """``source_window_days`` is optional: the 15-minute schedule omits it and
-        gets the task's default, the daily reconciliation pass widens it.
-        """
         body = request.data if isinstance(request.data, dict) else {}
-        if "source_window_days" not in body:
-            return self._run(aggregate_metrics_from_sources)
+        kwargs: dict[str, Any] = {}
         try:
-            days = _int_arg(request, "source_window_days", DASHBOARD_SOURCE_WINDOW_DAYS)
+            if body.get("tier") is not None:
+                kwargs["tier"] = _tier_arg(body["tier"])
+            if body.get("source_window_days") is not None:
+                kwargs["source_window_days"] = _int_arg(
+                    request, "source_window_days", DASHBOARD_SOURCE_WINDOW_DAYS
+                )
         except ValueError as exc:
+            # The one branch _run no longer covers, so it is logged here or nowhere.
+            logger.warning("dashboard-metrics aggregate rejected: %s", exc)
             return Response({"error": str(exc)}, status=status.HTTP_400_BAD_REQUEST)
-        return self._run(aggregate_metrics_from_sources, source_window_days=days)
+        return self._run(aggregate_metrics_from_sources, **kwargs)
 
 
 class CleanupHourlyMetricsAPIView(_MetricsTaskAPIView):

backend/dashboard_metrics/migrations/0006_split_aggregation_schedule.py

@@ -0,0 +1,212 @@
+"""Split the metrics aggregation into two schedules by tier (UN-3974).
+
+The hourly tier keeps its 15-minute cadence; the daily and monthly tiers move to
+hourly, taking the expensive DAY-granularity half of the work from 96 runs a day to
+24. Both rows run the same task and differ only in their ``tier`` kwargs — a second
+task name would need its own worker registration and internal endpoint.
+
+Beat and PG rows are declared here from one spec, so this pair cannot drift the way
+0002 and 0004 can. Beat stores kwargs as a JSON string, PgPeriodicTask decoded. The
+new row inherits whichever scheduler owns the row it is split from, rather than
+hardcoding Beat: it is one half of that row, and the same process should fire it.
+
+The new row runs at minute 20 — off the ``*/15`` grid — so it never starts alongside
+the hourly-tier run, whose per-tier lock is deliberately unable to block it.
+
+**Rolling back the code past this release requires reversing this migration too.**
+After it runs both scheduler rows carry a ``tier`` kwarg that the previous release's
+zero-argument signatures reject with ``TypeError``, which ``autoretry_for`` does not
+cover — aggregation would stop for both tiers until ``migrate dashboard_metrics 0005``
+restores the single row.
+"""
+
+import json
+
+from django.db import migrations
+from django.utils import timezone
+
+AGGREGATE_TASK_NAME = "dashboard_metrics.aggregate_from_sources"
+AGGREGATE_QUEUE = "dashboard_metric_events"
+
+# Frozen wire values, not a copy to keep in step with the enum. These are written
+# into rows this migration never re-runs against, so editing them to follow a rename
+# of AggregationTier changes nothing in production and every test stays green while
+# live rows still carry the old string — the task then raises ValueError on every
+# run. Renaming an AggregationTier value needs a NEW data migration that rewrites the
+# rows; this file is a record of what was written on the day it ran.
+TIER_HOURLY = "hourly"
+TIER_DAILY_MONTHLY = "daily_monthly"
+
+# Created by 0002 / 0004; only its kwargs and description change here.
+EXISTING_AGGREGATE_ROW = "dashboard_metrics_aggregate_from_sources"
+
+AGGREGATION_SCHEDULES = [
+    {
+        "name": EXISTING_AGGREGATE_ROW,
+        "tier": TIER_HOURLY,
+        "cron_string": "*/15 * * * *",
+        "crontab": {"minute": "*/15", "hour": "*"},
+        "description": (
+            "Aggregate the hourly dashboard metrics tier from source tables "
+            "(Usage, PageUsage, WorkflowExecution, etc.)"
+        ),
+        "exists": True,
+    },
+    {
+        "name": "dashboard_metrics_aggregate_daily_monthly",
+        "tier": TIER_DAILY_MONTHLY,
+        # Off the */15 grid (:00 :15 :30 :45): the per-tier locks are built so the
+        # two runs cannot block each other, so a shared minute means two full
+        # prefilter scans and two per-org loops at once. Same cadence, no overlap.
+        "cron_string": "20 * * * *",
+        "crontab": {"minute": "20", "hour": "*"},
+        "description": (
+            "Aggregate the daily and monthly dashboard metrics tiers from source "
+            "tables — hourly, since these figures do not need 15-minute freshness"
+        ),
+        "exists": False,
+    },
+]
+
+
+def _inherited_ownership(periodic_task_model, pg_periodic_task_model):
+    """Which scheduler fires the row being split, so its other half matches.
+
+    Hardcoding Beat would leave the daily/monthly tier with no firer wherever the
+    metrics periodics are already PG-adopted: the adopted row's Beat twin is disabled
+    and Beat may not be running at all.
+    """
+    beat = periodic_task_model.objects.filter(name=EXISTING_AGGREGATE_ROW).first()
+    pg = pg_periodic_task_model.objects.filter(name=EXISTING_AGGREGATE_ROW).first()
+    return {
+        "beat_enabled": True if beat is None else beat.enabled,
+        "pg_enabled": True if pg is None else pg.enabled,
+        "pg_owned": False if pg is None else pg.pg_owned,
+    }
+
+
+def split_schedules(apps, schema_editor):
+    CrontabSchedule = apps.get_model("django_celery_beat", "CrontabSchedule")
+    PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask")
+    PgPeriodicTask = apps.get_model("pg_queue", "PgPeriodicTask")
+    owner = _inherited_ownership(PeriodicTask, PgPeriodicTask)
+
+    for spec in AGGREGATION_SCHEDULES:
+        kwargs = {"tier": spec["tier"]}
+
+        if spec["exists"]:
+            # Payload only. `enabled` and `pg_owned` say which scheduler fires this
+            # row and belong to converge_pg_scheduler; rewriting them here can leave
+            # an adopted row with no firer. Its cadence does not change.
+            #
+            # The counts are checked rather than discarded: a bulk update matching no
+            # row reports success having changed nothing, leaving the old row on
+            # kwargs="{}" — which defaults to every tier every 15 minutes — while the
+            # new hourly row also fires. Strictly more load than before, silently.
+            beat_updated = PeriodicTask.objects.filter(name=spec["name"]).update(
+                kwargs=json.dumps(kwargs), description=spec["description"]
+            )
+            pg_updated = PgPeriodicTask.objects.filter(name=spec["name"]).update(
+                task_kwargs=kwargs
+            )
+            if not beat_updated or not pg_updated:
+                raise RuntimeError(
+                    f"{spec['name']}: expected a row on both schedulers to split, "
+                    f"found beat={beat_updated} pg={pg_updated}. Apply 0002 and 0004 "
+                    "first, or restore the row before re-running."
+                )
+            continue
+
+        schedule, _ = CrontabSchedule.objects.get_or_create(
+            minute=spec["crontab"]["minute"],
+            hour=spec["crontab"]["hour"],
+            day_of_week="*",
+            day_of_month="*",
+            month_of_year="*",
+            defaults={"timezone": "UTC"},
+        )
+        PeriodicTask.objects.update_or_create(
+            name=spec["name"],
+            defaults={
+                "task": AGGREGATE_TASK_NAME,
+                "crontab": schedule,
+                "queue": AGGREGATE_QUEUE,
+                "kwargs": json.dumps(kwargs),
+                "enabled": owner["beat_enabled"],
+                "description": spec["description"],
+            },
+        )
+        PgPeriodicTask.objects.update_or_create(
+            name=spec["name"],
+            defaults={
+                "task_name": AGGREGATE_TASK_NAME,
+                "queue": AGGREGATE_QUEUE,
+                "task_args": [],
+                "task_kwargs": kwargs,
+                "cron_string": spec["cron_string"],
+                "org_id": "",
+                "enabled": owner["pg_enabled"],
+                "pg_owned": owner["pg_owned"],
+            },
+        )
+
+    _bump_beat_change_tracker(apps)
+
+
+def merge_schedules(apps, schema_editor):
+    """Restore the single every-15-minutes row that writes all three tiers.
+
+    Leaves `enabled` / `pg_owned` alone, as the forward direction does.
+    """
+    PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask")
+    PgPeriodicTask = apps.get_model("pg_queue", "PgPeriodicTask")
+
+    added = [s["name"] for s in AGGREGATION_SCHEDULES if not s["exists"]]
+    PeriodicTask.objects.filter(name__in=added).delete()
+    PgPeriodicTask.objects.filter(name__in=added).delete()
+
+    beat_restored = PeriodicTask.objects.filter(name=EXISTING_AGGREGATE_ROW).update(
+        kwargs="{}",
+        description=(
+            "Aggregate metrics from source tables (Usage, PageUsage, etc.) "
+            "into hourly, daily, and monthly metrics tables"
+        ),
+    )
+    pg_restored = PgPeriodicTask.objects.filter(name=EXISTING_AGGREGATE_ROW).update(
+        task_kwargs={}
+    )
+    if not beat_restored or not pg_restored:
+        raise RuntimeError(
+            f"{EXISTING_AGGREGATE_ROW}: expected a row on both schedulers to restore, "
+            f"found beat={beat_restored} pg={pg_restored}. The rollback would leave "
+            "the daily and monthly tiers with no schedule."
+        )
+    _bump_beat_change_tracker(apps)
+
+
+def _bump_beat_change_tracker(apps):
+    """Make a running Beat reload instead of keeping the pre-split schedule.
+
+    django-celery-beat's post_save receiver binds the concrete PeriodicTask, so writes
+    through a historical model never bump PeriodicTasks.last_update and
+    DatabaseScheduler keeps its in-memory copy: the existing row would go on firing
+    with no tier and the new row would never fire at all — the whole saving silently
+    not happening. Same fix and reason as scheduler/ownership.py and
+    mirror_pg_periodic_tasks.py.
+    """
+    periodic_tasks_model = apps.get_model("django_celery_beat", "PeriodicTasks")
+    periodic_tasks_model.objects.update_or_create(
+        ident=1, defaults={"last_update": timezone.now()}
+    )
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("dashboard_metrics", "0005_add_reconciliation_task"),
+        ("django_celery_beat", "0018_improve_crontab_helptext"),
+        ("pg_queue", "0003_pgperiodictask"),
+    ]
+
+    operations = [
+        migrations.RunPython(split_schedules, merge_schedules),
+    ]