UN-3883 [FEAT] Cut dashboard metrics cron DB load: narrower source windows, monthly from daily, two new indexes

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,7 +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/monthly |
+| `aggregate_from_sources` | Every 15 min | Aggregates source → **hourly tier only** (`tier=hourly`) |
+| `aggregate_from_sources` (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 |
 
@@ -109,7 +111,7 @@ celery -A backend beat -l info
 │ EventMetrics    │      │ EventMetrics    │      │ EventMetrics    │
 │ Hourly          │      │ Daily           │      │ Monthly         │
 │                 │      │                 │      │                 │
-│ • 24h query     │      │ • 7 day query   │      │ • 2 month query │
+│ • 24h query     │      │ • 2 day query   │      │ • from daily    │
 │ • 30 day retain │      │ • 365 day retain│      │ • No cleanup    │
 └────────┬────────┘      └────────┬────────┘      └────────┬────────┘
          │                        │                        │
@@ -162,7 +164,9 @@ 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 **daily- or monthly-tier** gap shorter than that repairs itself without a manual backfill. The hourly tier always covers only the last 24h, so an hourly gap needs `backfill_metrics` regardless.
+- The 7-day window is also the ceiling on lag, not just on downtime. The source queries filter on a terminal status but window and bucket on `created_at`, so a row whose status turns terminal more than 7 days after it was created is counted in no daily row — and therefore in no monthly total either, since monthly is the sum of daily. Before the monthly tier was derived from daily this was caught by the wider monthly source window.
 - Celery tasks have `max_retries=3` with exponential backoff.
 - Cleanup tasks (hourly: 30-day retention, daily: 365-day retention) prevent unbounded table growth.
 
@@ -298,8 +302,8 @@ cost = (input_cost_per_token × input_tokens) + (output_cost_per_token × output
 | Table | Model | Time Column | Granularity | Query Window | Retention |
 |-------|-------|-------------|-------------|--------------|-----------|
 | `event_metrics_hourly` | `EventMetricsHourly` | `timestamp` | Hour | Last 24 hours | 30 days |
-| `event_metrics_daily` | `EventMetricsDaily` | `date` | Day | Last 7 days | 365 days |
-| `event_metrics_monthly` | `EventMetricsMonthly` | `month` | Month | Last 2 months | Forever |
+| `event_metrics_daily` | `EventMetricsDaily` | `date` | Day | Last 2 days (7 on the daily reconciliation pass) | 365 days |
+| `event_metrics_monthly` | `EventMetricsMonthly` | `month` | Month | Rolled up from the daily tier, current + previous month | Forever |
 
 ### Table Schema
 
@@ -339,7 +343,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` | 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 |
 
@@ -378,16 +384,38 @@ The `aggregate_metrics_from_sources` task:
 2. **For each metric**:
    - Queries source table with `MetricsQueryService`
    - Groups by time period (hour/day/month)
-3. **Upserts results** into aggregated tables using `update_or_create`
-4. **Uses `_base_manager`** to bypass Django's organization filter in Celery context
+3. **Upserts results** into the hourly and daily tables
+4. **Rolls monthly up from the daily tier** in one statement for all orgs. Upsert-only:
+   a monthly row the daily tier no longer produces is left in place. A stale total is
+   recoverable with `backfill_metrics`; a deleted one is not, because the daily rows
+   that would rebuild it are exactly what is missing
+5. **Uses `_base_manager`** to bypass Django's organization filter in Celery context
 
 ```python
 # Query windows
-hourly_start = end_date - timedelta(hours=24)    # Last 24 hours
-daily_start = end_date - timedelta(days=7)       # Last 7 days
-monthly_start = first_of_previous_month          # Last 2 months
+hourly_start = end_date - timedelta(hours=24)                 # Last 24 hours
+daily_start = truncate_to_day(end_date - source_window_days)  # 2 days, 7 on reconcile
+monthly_start = first_of_previous_month                       # summed from daily
 ```
 
+The monthly tier has no source queries of its own. `backfill_metrics` still computes
+monthly from source, so within the rollup window (current + previous month) its output
+is overwritten by the sum of the daily tier on the next daily/monthly pass — see that
+command's help text. **Backfill daily before relying on monthly:** the rollup writes
+whatever daily holds, so a month whose daily tier is short produces an under-counted
+monthly total.
+
+Run this once before the first aggregation after deploying the monthly-from-daily
+rollup, so the tier it derives from is complete:
+
+```
+python manage.py backfill_metrics --days 62 --skip-hourly --skip-monthly
+```
+
+62, not 60: the rollup window reaches back to the first of the previous month, which is
+61 days before a run on the 31st. `--skip-monthly` is deliberate — repair daily and let
+the rollup derive monthly from it.
+
 ---
 
 ## API Endpoints

backend/dashboard_metrics/internal_views.py

@@ -34,6 +34,9 @@
 from utils.local_context import StateStore
 
 from dashboard_metrics.tasks import (
+    DASHBOARD_SOURCE_WINDOW_DAYS,
+    MAX_SOURCE_WINDOW_DAYS,
+    AggregationTier,
     aggregate_metrics_from_sources,
     cleanup_daily_metrics,
     cleanup_hourly_metrics,
@@ -58,7 +61,16 @@ def _clear_org_context() -> None:
         StateStore.clear(Account.ORGANIZATION_ID)
 
 
-def _int_arg(request: Request, key: str, default: int) -> int:
+def _tier_arg(raw: Any) -> AggregationTier:
+    """Coerce a request body's tier to the enum, raising ValueError on anything else."""
+    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, maximum: int | None = None) -> int:
     """Read an optional positive integer from the request body."""
     raw = request.data.get(key, default) if isinstance(request.data, dict) else default
     try:
@@ -67,18 +79,21 @@ def _int_arg(request: Request, key: str, default: int) -> int:
         raise ValueError(f"{key} must be an integer, got {raw!r}") from exc
     if value < 1:
         raise ValueError(f"{key} must be >= 1, got {value}")
+    if maximum is not None and value > maximum:
+        raise ValueError(f"{key} must be <= {maximum}, got {value}")
     return value
 
 
 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(
@@ -91,10 +106,32 @@ 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:
-        return self._run(aggregate_metrics_from_sources)
+        body = request.data if isinstance(request.data, dict) else {}
+        kwargs: dict[str, Any] = {}
+        try:
+            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,
+                    maximum=MAX_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, **kwargs)
 
 
 class CleanupHourlyMetricsAPIView(_MetricsTaskAPIView):

backend/dashboard_metrics/management/commands/backfill_metrics.py

@@ -3,6 +3,12 @@
 This command populates EventMetricsHourly, EventMetricsDaily, and EventMetricsMonthly
 tables from historical data in source tables (Usage, PageUsage, WorkflowExecution, etc.)
 
+The current and previous month are recomputed from the daily tier by the aggregation
+task's daily/monthly pass, so inside that window this command's monthly output is
+overwritten and --skip-monthly is a no-op. --skip-daily is worse than useless there:
+monthly is rebuilt from a tier this run did not populate, producing an under-count.
+Backfill both, or neither.
+
 Usage:
     python manage.py backfill_metrics --days=30
     python manage.py backfill_metrics --days=90 --org-id=5
@@ -26,6 +32,7 @@
     MetricType,
 )
 from dashboard_metrics.services import MetricsQueryService
+from dashboard_metrics.tasks import _truncate_to_day
 
 logger = logging.getLogger(__name__)
 
@@ -92,12 +99,19 @@ def add_arguments(self, parser):
         parser.add_argument(
             "--skip-daily",
             action="store_true",
-            help="Skip daily aggregation",
+            help=(
+                "Skip daily aggregation. Unsafe for the current and previous month: "
+                "the aggregation task rebuilds monthly from daily there, so monthly "
+                "ends up under-counted."
+            ),
         )
         parser.add_argument(
             "--skip-monthly",
             action="store_true",
-            help="Skip monthly aggregation",
+            help=(
+                "Skip monthly aggregation. A no-op for the current and previous "
+                "month, which the aggregation task owns."
+            ),
         )
         parser.add_argument(
             "--active-only",
@@ -118,11 +132,23 @@ def handle(self, *args, **options):
         active_only = options["active_only"]
 
         end_date = timezone.now()
-        start_date = end_date - timedelta(days=days)
+        # Truncated to match the cron's daily_start: an untruncated boundary writes the
+        # oldest day covering only part of it, and the monthly rollup now sums the
+        # persisted daily tier rather than recomputing that day from source.
+        start_date = _truncate_to_day(end_date - timedelta(days=days))
 
         self.stdout.write(f"Backfill period: {start_date.date()} to {end_date.date()}")
         self.stdout.write(f"Days: {days}")
 
+        if skip_daily and not skip_monthly:
+            self.stdout.write(
+                self.style.WARNING(
+                    "--skip-daily without --skip-monthly: the aggregation task "
+                    "rebuilds the current and previous month from the daily tier, "
+                    "so monthly will be overwritten with an under-count."
+                )
+            )
+
         if dry_run:
             self.stdout.write(self.style.WARNING("DRY RUN - no changes will be made"))
 

backend/dashboard_metrics/migrations/0005_add_reconciliation_task.py

@@ -0,0 +1,125 @@
+"""Data migration to schedule the daily-tier reconciliation pass.
+
+The 15-minute aggregation reads a narrow source window, which cannot repair
+gaps left by cron downtime. This runs the same task once a day at a wider
+window to backfill them.
+
+Declared for **both** transports, like 0002/0004: Beat reads
+``django_celery_beat_periodictask``, the PG scheduler reads ``pg_periodic_task``,
+and a schedule present on one only stops firing the moment the flag flips.
+``kwargs`` is a JSON string on Beat and a JSONField on PG — same value, two
+encodings.
+
+The row carries ``source_window_days``, which the previous release's zero-argument
+signatures reject with ``TypeError``. Rolling the code back past this release means
+reversing this migration too, **before** the image rolls back — ``migrate
+dashboard_metrics 0004``, which reverses 0006 and this one together.
+"""
+
+from django.db import migrations
+from django.utils import timezone
+
+RECONCILE_TASK_NAME = "dashboard_metrics_reconcile_source_window"
+RECONCILE_DESCRIPTION = (
+    "Re-aggregate metrics over a 7 day source window to repair "
+    "daily-tier gaps left by cron downtime"
+)
+
+# Single source for both directions, and importable by the drift test.
+PG_PERIODIC_TASKS = [
+    {
+        "name": RECONCILE_TASK_NAME,
+        "task_name": "dashboard_metrics.aggregate_from_sources",
+        "queue": "dashboard_metric_events",
+        "task_args": [],
+        "task_kwargs": {"source_window_days": 7},
+        # Beat: CrontabSchedule(minute=40, hour=4, every day) UTC — clear of the
+        # 2:00 and 3:00 cleanup tasks, and off the aggregation's */15 grid
+        # (:00 :15 :30 :45) so the two never start together.
+        "cron_string": "40 4 * * *",
+    },
+]
+
+
+def create_reconciliation_task(apps, schema_editor):
+    """Create the once-daily reconciliation periodic task on both transports."""
+    crontab_model = apps.get_model("django_celery_beat", "CrontabSchedule")
+    periodic_task_model = apps.get_model("django_celery_beat", "PeriodicTask")
+    pg_periodic_task_model = apps.get_model("pg_queue", "PgPeriodicTask")
+
+    schedule_4am, _ = crontab_model.objects.get_or_create(
+        minute="40",
+        hour="4",
+        day_of_week="*",
+        day_of_month="*",
+        month_of_year="*",
+        defaults={"timezone": "UTC"},
+    )
+
+    for spec in PG_PERIODIC_TASKS:
+        periodic_task_model.objects.update_or_create(
+            name=spec["name"],
+            defaults={
+                "task": spec["task_name"],
+                "crontab": schedule_4am,
+                "queue": spec["queue"],
+                "kwargs": '{"source_window_days": 7}',
+                "enabled": True,
+                "description": RECONCILE_DESCRIPTION,
+            },
+        )
+        pg_periodic_task_model.objects.update_or_create(
+            name=spec["name"],
+            defaults={
+                "task_name": spec["task_name"],
+                "queue": spec["queue"],
+                "task_args": spec["task_args"],
+                "task_kwargs": spec["task_kwargs"],
+                "cron_string": spec["cron_string"],
+                "org_id": "",
+                "enabled": True,
+                # Inert until the rollout flag decides otherwise.
+                "pg_owned": False,
+            },
+        )
+
+    _bump_beat_change_tracker(apps)
+
+
+def remove_reconciliation_task(apps, schema_editor):
+    """Remove the reconciliation periodic task from both transports."""
+    names = [spec["name"] for spec in PG_PERIODIC_TASKS]
+    apps.get_model("django_celery_beat", "PeriodicTask").objects.filter(
+        name__in=names
+    ).delete()
+    apps.get_model("pg_queue", "PgPeriodicTask").objects.filter(name__in=names).delete()
+    _bump_beat_change_tracker(apps)
+
+
+def _bump_beat_change_tracker(apps):
+    """Make a running Beat reload instead of missing the new 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 stale in-memory copy. 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", "0004_pg_periodic_tasks"),
+        ("django_celery_beat", "0018_improve_crontab_helptext"),
+        ("pg_queue", "0003_pgperiodictask"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            create_reconciliation_task,
+            remove_reconciliation_task,
+        ),
+    ]