diff --git a/backend/dashboard_metrics/README.md b/backend/dashboard_metrics/README.md index 8c3fee6ce8..cd7aab29b9 100644 --- a/backend/dashboard_metrics/README.md +++ b/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 | diff --git a/backend/dashboard_metrics/internal_views.py b/backend/dashboard_metrics/internal_views.py index 0ecc6759f2..fce4fc978d 100644 --- a/backend/dashboard_metrics/internal_views.py +++ b/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): diff --git a/backend/dashboard_metrics/migrations/0006_split_aggregation_schedule.py b/backend/dashboard_metrics/migrations/0006_split_aggregation_schedule.py new file mode 100644 index 0000000000..21fda1bff2 --- /dev/null +++ b/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), + ] diff --git a/backend/dashboard_metrics/tasks.py b/backend/dashboard_metrics/tasks.py index f9b896f68a..691b51b490 100644 --- a/backend/dashboard_metrics/tasks.py +++ b/backend/dashboard_metrics/tasks.py @@ -9,6 +9,7 @@ import logging import time from datetime import date, datetime, timedelta +from enum import StrEnum from typing import Any from account_v2.models import Organization @@ -237,20 +238,80 @@ def _rollup_monthly_from_daily(month_start: date) -> int: return len(objects) +class AggregationTier(StrEnum): + """Which metric tiers one aggregation run writes. + + Daily and monthly stay together because monthly is rolled up from the daily tier. + """ + + HOURLY = "hourly" + DAILY_MONTHLY = "daily_monthly" + ALL = "all" + + +# Which granularities each tier writes. One table rather than a predicate per +# granularity: add a member without an entry here and _tiers_written raises on the +# first run, instead of the run acquiring its lock, iterating every org, writing +# nothing and returning success. +_TIER_WRITES: dict[AggregationTier, frozenset[str]] = { + AggregationTier.HOURLY: frozenset({AggregationTier.HOURLY.value}), + AggregationTier.DAILY_MONTHLY: frozenset({AggregationTier.DAILY_MONTHLY.value}), + AggregationTier.ALL: frozenset( + {AggregationTier.HOURLY.value, AggregationTier.DAILY_MONTHLY.value} + ), +} + + +def _tiers_written(tier: AggregationTier) -> frozenset[str]: + """The granularities one tier writes. Unhandled members raise rather than no-op.""" + try: + return _TIER_WRITES[tier] + except KeyError: + raise AssertionError(f"Unhandled AggregationTier: {tier!r}") from None + + +def _writes_hourly(tier: AggregationTier) -> bool: + return AggregationTier.HOURLY.value in _tiers_written(tier) + + +def _writes_daily_monthly(tier: AggregationTier) -> bool: + return AggregationTier.DAILY_MONTHLY.value in _tiers_written(tier) + + AGGREGATION_LOCK_KEY_PREFIX = "dashboard_metrics:aggregation_lock" AGGREGATION_LOCK_TIMEOUT = 900 # 15 minutes (matches task schedule) -def _aggregation_lock_key(source_window_days: int) -> str: - """One key per schedule, not one key for the task. +def _aggregation_lock_keys(tier: AggregationTier, source_window_days: int) -> list[str]: + """One key per granularity written, namespaced by source window. - The 15-minute row is an IntervalSchedule and drifts against the reconciliation - pass's fixed crontab, so on a shared key the once-daily repair would lose the - race and return skipped=True — never retried, and the only mechanism that - repairs the narrowed window. Distinct windows are distinct jobs; both are - idempotent upserts, so the once-a-day overlap costs duplicated work at worst. + Per granularity, not per enum member: keying on the label alone gives ALL a third + key that excludes nothing, so an ALL run and the scheduled hourly run would write + EventMetricsHourly concurrently. Taking one key per granularity restores exclusion + exactly where writes collide, and the two scheduled tiers still never block. + + Per window because a wider window is a different job. The reconciliation pass runs + once a day on a fixed crontab against a drifting 15-minute interval; on a shared + key it would lose the race, return skipped=True and never be retried — and it is + the only thing that repairs the narrowed window. Both are idempotent upserts, so + that once-a-day overlap costs duplicated work at worst. """ - return f"{AGGREGATION_LOCK_KEY_PREFIX}:{source_window_days}d" + return [ + f"{AGGREGATION_LOCK_KEY_PREFIX}:{source_window_days}d:{granularity}" + for granularity in sorted(_tiers_written(tier)) + ] + + +def _acquire_aggregation_locks(lock_keys: list[str]) -> list[str]: + """Take every key or none; returns the keys taken, empty if the run must skip.""" + taken: list[str] = [] + for key in lock_keys: + if not _acquire_aggregation_lock(key): + for held in taken: + cache.delete(held) + return [] + taken.append(key) + return taken def _acquire_aggregation_lock(lock_key: str) -> bool: @@ -306,41 +367,56 @@ def _acquire_aggregation_lock(lock_key: str) -> bool: retry_backoff_max=300, ) def aggregate_metrics_from_sources( + tier: str = AggregationTier.ALL, source_window_days: int = DASHBOARD_SOURCE_WINDOW_DAYS, ) -> dict[str, Any]: """Aggregate source tables into the hourly, daily and monthly tiers. - Runs every 15 minutes under a self-healing Redis lock. Hourly covers the - last 24h, daily the source window, monthly is rolled up from daily. + Three schedules call this: the hourly tier every 15 minutes, the daily and + monthly tiers hourly at :20, and a once-daily reconciliation pass over every + tier at a wider window. Hourly covers the last 24h, daily the source window, + monthly is rolled up from daily. Args: - source_window_days: Daily-tier source lookback. The once-daily - reconciliation pass reruns this task at - DASHBOARD_RECONCILE_WINDOW_DAYS to repair gaps after downtime. + tier: An AggregationTier value. Defaults to all, so a caller that omits it + — a schedule row written before 0006 — writes every tier rather than + none. + source_window_days: Daily-tier source lookback. The reconciliation pass + reruns this task at DASHBOARD_RECONCILE_WINDOW_DAYS to repair gaps + after downtime. Returns: - Dict with aggregation summary for all three tiers + Dict with aggregation summary for the tiers that ran + + Raises: + ValueError: tier is not a recognised AggregationTier, or the window is not + an integer between 1 and MAX_SOURCE_WINDOW_DAYS """ + tier = AggregationTier(tier) source_window_days = _validate_source_window(source_window_days) - lock_key = _aggregation_lock_key(source_window_days) + lock_keys = _aggregation_lock_keys(tier, source_window_days) - if not _acquire_aggregation_lock(lock_key): + held = _acquire_aggregation_locks(lock_keys) + if not held: logger.warning( - "Skipping the %d-day aggregation — another run of the same schedule is " - "in progress", + "Skipping the %s aggregation over %d day(s) — another run writing the " + "same tier is in progress", + tier.value, source_window_days, ) return { "success": True, "skipped": True, "reason": "lock_held", + "tier": tier.value, "source_window_days": source_window_days, } try: - return _run_aggregation(source_window_days) + return _run_aggregation(tier, source_window_days) finally: - cache.delete(lock_key) + for key in held: + cache.delete(key) def _aggregate_single_metric( @@ -353,24 +429,29 @@ def _aggregate_single_metric( end_date: datetime, hourly_agg: dict, daily_agg: dict, + tier: AggregationTier, extra_kwargs: dict | None = None, ) -> None: - """Run a single metric query at hourly and daily granularity.""" + """Run a single metric query at the granularities this run writes.""" extra_kwargs = extra_kwargs or {} # === HOURLY (last 24h) === - for row in query_method( - org_id, - hourly_start, - end_date, - granularity=Granularity.HOUR, - **extra_kwargs, - ): - hour_ts = _truncate_to_hour(row["period"]) - key = (org_id, hour_ts.isoformat(), metric_name, "default", "") - _upsert_agg(hourly_agg, key, metric_type, row["value"] or 0) + if _writes_hourly(tier): + for row in query_method( + org_id, + hourly_start, + end_date, + granularity=Granularity.HOUR, + **extra_kwargs, + ): + hour_ts = _truncate_to_hour(row["period"]) + key = (org_id, hour_ts.isoformat(), metric_name, "default", "") + _upsert_agg(hourly_agg, key, metric_type, row["value"] or 0) + + # === DAILY (monthly is rolled up from it, so one query feeds both) === + if not _writes_daily_monthly(tier): + return - # === DAILY === for row in query_method( org_id, daily_start, @@ -391,24 +472,29 @@ def _aggregate_llm_combined( hourly_agg: dict, daily_agg: dict, llm_combined_fields: dict, + tier: AggregationTier, ) -> None: - """Run the combined LLM metrics query at hourly and daily granularity. + """Run the combined LLM metrics query at the granularities this run writes. Two queries covering four metrics. """ # === HOURLY (last 24h) === - for row in MetricsQueryService.get_llm_metrics_combined( - org_id, - hourly_start, - end_date, - granularity=Granularity.HOUR, - ): - ts_str = _truncate_to_hour(row["period"]).isoformat() - for field, (metric_name, metric_type) in llm_combined_fields.items(): - key = (org_id, ts_str, metric_name, "default", "") - _upsert_agg(hourly_agg, key, metric_type, row[field] or 0) + if _writes_hourly(tier): + for row in MetricsQueryService.get_llm_metrics_combined( + org_id, + hourly_start, + end_date, + granularity=Granularity.HOUR, + ): + ts_str = _truncate_to_hour(row["period"]).isoformat() + for field, (metric_name, metric_type) in llm_combined_fields.items(): + key = (org_id, ts_str, metric_name, "default", "") + _upsert_agg(hourly_agg, key, metric_type, row[field] or 0) # === DAILY === + if not _writes_daily_monthly(tier): + return + for row in MetricsQueryService.get_llm_metrics_combined( org_id, daily_start, @@ -450,6 +536,7 @@ def _collect_org_metrics( hourly_start: datetime, daily_start: datetime, end_date: datetime, + tier: AggregationTier, ) -> tuple[dict, dict, int]: """Query every metric for one org into hourly/daily aggregates. @@ -482,6 +569,7 @@ def _collect_org_metrics( end_date, hourly_agg, daily_agg, + tier, extra_kwargs, ) except Exception: @@ -497,6 +585,7 @@ def _collect_org_metrics( hourly_agg, daily_agg, LLM_COMBINED_FIELDS, + tier, ) except Exception: logger.exception("Error querying combined LLM metrics for org %s", org_id) @@ -510,11 +599,12 @@ def _aggregate_org( hourly_start: datetime, daily_start: datetime, end_date: datetime, + tier: AggregationTier, stats: dict[str, Any], ) -> None: - """Aggregate one organization and upsert its hourly and daily tiers.""" + """Aggregate one organization and upsert the tiers this run writes.""" hourly_agg, daily_agg, errors = _collect_org_metrics( - org, hourly_start, daily_start, end_date + org, hourly_start, daily_start, end_date, tier ) stats["errors"] += errors @@ -551,10 +641,12 @@ def _build_result( daily_start: datetime, monthly_start: date, end_date: datetime, + tier: AggregationTier, skipped_reason: str | None = None, ) -> dict[str, Any]: """Shape the task's return value from the accumulated stats.""" result = { + "tier": tier.value, # Not a literal: every metric for every org can fail while each exception is # caught per-metric, and the run would otherwise report 200 / success with # zero rows written and the dashboard frozen. @@ -597,10 +689,28 @@ def _validate_source_window(source_window_days: int) -> int: return days +def _roll_up_monthly(monthly_start: date, stats: dict[str, Any]) -> None: + """Derive the monthly tier from daily, recording a failure distinctly.""" + try: + stats["monthly"]["upserted"] = _rollup_monthly_from_daily(monthly_start) + except (DatabaseError, OperationalError): + # Configured on the task for autoretry — swallowing them here would + # leave monthly permanently stale behind successful-looking runs. + raise + except Exception: + # upserted stays 0, which is also the legitimate empty-rollup value, so + # mark the failure explicitly rather than letting the two collapse. + logger.exception("Error rolling up monthly metrics from %s", monthly_start) + stats["monthly"]["failed"] = True + stats["errors"] += 1 + + def _run_aggregation( + tier: AggregationTier = AggregationTier.ALL, source_window_days: int = DASHBOARD_SOURCE_WINDOW_DAYS, ) -> dict[str, Any]: """Execute the aggregation, separately from the task's lock handling.""" + tier = AggregationTier(tier) source_window_days = _validate_source_window(source_window_days) end_date = timezone.now() @@ -621,11 +731,9 @@ def _run_aggregation( # Pre-filter to orgs with recent activity to reduce DB load. active_org_ids = _active_org_ids(end_date, daily_start) - logger.info( - "Aggregation: %d active orgs out of %d total", - len(active_org_ids), - Organization.objects.count(), - ) + # No total_orgs here: a full count of the organization table, on every run of + # every tier, whose only consumer was this log line. + logger.info("Aggregation (%s): %d active orgs", tier.value, len(active_org_ids)) if not active_org_ids: return _build_result( @@ -634,6 +742,7 @@ def _run_aggregation( daily_start, monthly_start, end_date, + tier, skipped_reason="no_active_orgs", ) @@ -643,23 +752,13 @@ def _run_aggregation( for org in organizations: try: - _aggregate_org(org, hourly_start, daily_start, end_date, stats) + _aggregate_org(org, hourly_start, daily_start, end_date, tier, stats) except Exception: logger.exception("Error processing org %s", org.id) stats["errors"] += 1 - try: - stats["monthly"]["upserted"] = _rollup_monthly_from_daily(monthly_start) - except (DatabaseError, OperationalError): - # Configured on the task for autoretry — swallowing them here would - # leave monthly permanently stale behind successful-looking runs. - raise - except Exception: - # upserted stays 0, which is also the legitimate empty-rollup value, so - # mark the failure explicitly rather than letting the two collapse. - logger.exception("Error rolling up monthly metrics from %s", monthly_start) - stats["monthly"]["failed"] = True - stats["errors"] += 1 + if _writes_daily_monthly(tier): + _roll_up_monthly(monthly_start, stats) log = logger.warning if stats["errors"] else logger.info log( @@ -670,7 +769,7 @@ def _run_aggregation( f"errors={stats['errors']}" ) - return _build_result(stats, hourly_start, daily_start, monthly_start, end_date) + return _build_result(stats, hourly_start, daily_start, monthly_start, end_date, tier) @shared_task( diff --git a/backend/dashboard_metrics/tests/test_active_org_prefilter.py b/backend/dashboard_metrics/tests/test_active_org_prefilter.py new file mode 100644 index 0000000000..d7893c653d --- /dev/null +++ b/backend/dashboard_metrics/tests/test_active_org_prefilter.py @@ -0,0 +1,141 @@ +"""The active-org prefilter can actually use we_created_at_idx (UN-3974, AC-3). + +AC-3 is worded as a production observation — "no longer appears in the top 10 by total +execution time in Query Insights" — and that half can only be read off production. The +half that is answerable here is the one underneath it: the prefilter bounds nothing but +`created_at`, and the index exists to serve exactly that shape. + +What this pins is the pairing. `workflow_manager/workflow_v2/tests/test_we_created_at_idx.py` +proves the index is declared and built safely; this proves the query still looks like +something it can serve. Either half can drift without the other noticing — someone +narrowing the prefilter to lead with a different column leaves the index built, valid, +and dead. + +Rows are inserted in ascending `created_at` order so the heap matches production, where +executions are appended as they happen. With them scattered the planner reads the whole +composite (workflow_id, created_at DESC) index instead, which is an artefact of the +fixture rather than anything about the query. + +**Not production evidence.** A few thousand rows in an otherwise-empty table on a +locally-configured Postgres is not the production planner's input: index-vs-seq-scan at +this selectivity is a cost-model output, sensitive to the PG major version, +`random_page_cost`, `effective_cache_size` and parallel workers, none of which are +pinned here. What the plan assertion below rules out is the *regression* — a prefilter +that has to read the executions table whatever the costs say. Whether production picks +the index is measured on production, and belongs to AC-3. + +DB-bound, so conftest marks it integration. +""" + +from __future__ import annotations + +import os + +import django +from django.apps import apps + +os.environ.setdefault("DJANGO_SETTINGS_MODULE", "backend.settings.test") +if not apps.ready: + django.setup() + +from account_v2.models import Organization # noqa: E402 +from django.db import connection # noqa: E402 +from django.test import TestCase # noqa: E402 +from django.test.utils import CaptureQueriesContext # noqa: E402 +from workflow_manager.workflow_v2.models.workflow import Workflow # noqa: E402 + +from dashboard_metrics.tasks import AggregationTier, _run_aggregation # noqa: E402 + +INDEX_NAME = "we_created_at_idx" + +_ROWS = 12000 +_SPAN_DAYS = 255 + + +class TestThePrefilterCanUseTheIndex(TestCase): + """Production ratios rather than production size: ~2.7% of rows in the 7-day window + is what decides whether the planner reaches for an index or scans. + """ + + def setUp(self) -> None: + self.org = Organization.objects.create( + organization_id="prefilter-org", name="prefilter", display_name="Prefilter" + ) + self.workflow = Workflow.objects.create( + workflow_name="prefilter-wf", organization=self.org + ) + with connection.cursor() as cur: + cur.execute( + """ + INSERT INTO workflow_execution ( + id, created_at, modified_at, workflow_id, execution_mode, + execution_method, execution_type, execution_log_id, status, + error_message, attempts, execution_time, result_acknowledged, + total_files) + SELECT gen_random_uuid(), ts, ts, %s, 'INSTANT', 'DIRECT', 'COMPLETE', + '', 'COMPLETED', '', 0, 1.0, false, 1 + FROM generate_series(1, %s) g + CROSS JOIN LATERAL ( + SELECT now() - (%s - (g::float / %s) * %s) * interval '1 day' + ) AS t(ts) + """, + [self.workflow.id, _ROWS, _SPAN_DAYS, _ROWS, _SPAN_DAYS], + ) + cur.execute("ANALYZE workflow_execution") + + def _prefilter_sql(self) -> str: + """The real query, taken from the task rather than rewritten here. + + A hand-copied queryset would keep passing after the prefilter changed, which is + the one thing this test is for. + """ + with CaptureQueriesContext(connection) as ctx: + _run_aggregation(AggregationTier.HOURLY) + candidates = [ + q["sql"] + for q in ctx.captured_queries + if "workflow_execution" in q["sql"] + and "DISTINCT" in q["sql"].upper() + and "created_at" in q["sql"] + ] + # The run issues nine further queries against this table, several of them + # joining it and filtering created_at. Index 0 is right today only by execution + # order, which nothing here states — so require the shape to be unambiguous. + assert candidates, "the aggregation issued no active-org prefilter query" + assert len(candidates) == 1, ( + f"{len(candidates)} queries match the prefilter shape; the match is no " + f"longer distinguishing:\n" + "\n\n".join(candidates) + ) + return str(candidates[0]) + + def test_the_window_is_the_selectivity_the_index_is_for(self) -> None: + """If the prefilter ever widened to most of the table, an index on created_at + would stop being the right answer — the planner would scan regardless. + """ + with connection.cursor() as cur: + cur.execute( + "SELECT count(*) FILTER (WHERE created_at >= now() - interval '7 days')" + "::float / count(*) FROM workflow_execution" + ) + share = cur.fetchone()[0] + assert 0 < share < 0.10 + + def test_the_index_can_serve_the_prefilter(self) -> None: + """*Usable*, not *chosen*. + + Whether the planner picks the index on a synthetic table turns on + random_page_cost, effective_cache_size, the PG major version and how the + freshly-loaded visibility map looks — none of which this fixture pins, so + asserting the choice reds the build on a config change with no code change. + Disabling seqscan asks the question that is actually about the query: can this + shape be served from the index at all? A prefilter narrowed to lead with a + different column fails here whatever the cost model says. + """ + sql = self._prefilter_sql() + with connection.cursor() as cur: + cur.execute("SET LOCAL enable_seqscan = off") + cur.execute("EXPLAIN " + sql) + plan = "\n".join(row[0] for row in cur.fetchall()) + assert f"Index Scan using {INDEX_NAME}" in plan or f"Index Only Scan using {INDEX_NAME}" in plan, ( + f"expected {INDEX_NAME} to be usable for the prefilter:\n{plan}" + ) diff --git a/backend/dashboard_metrics/tests/test_aggregation_dispatch.py b/backend/dashboard_metrics/tests/test_aggregation_dispatch.py new file mode 100644 index 0000000000..5a03e212b2 --- /dev/null +++ b/backend/dashboard_metrics/tests/test_aggregation_dispatch.py @@ -0,0 +1,141 @@ +"""Guard: the tier a schedule row declares reaches the task (UN-3974, AC-1). + +Two schedulers fire the same task name at two different implementations. Beat reads +``PeriodicTask.kwargs`` and calls the Django ``@shared_task`` directly; the PG scheduler +reads ``PgPeriodicTask.task_kwargs`` and goes through the worker proxy and the internal +endpoint to the same function. Both legs have to carry ``tier``, and a break in either is +invisible — the job still runs, still returns success, and just writes the wrong tiers. + +The worker half of the PG leg is pinned in ``workers/tests/test_dashboard_metrics_tasks.py``; +this covers the endpoint that receives it and the Beat leg's kwargs. + +DB-free: the task is mocked, and the Beat kwargs are read from the migration spec rather +than from a migrated database. +""" + +from __future__ import annotations + +import importlib +import inspect +import os +from typing import Any +from unittest import mock + +import django +import pytest +from django.apps import apps + +os.environ.setdefault("DJANGO_SETTINGS_MODULE", "backend.settings.test") +if not apps.ready: + django.setup() + +from rest_framework.test import APIRequestFactory # noqa: E402 + +from dashboard_metrics import internal_views # noqa: E402 +from dashboard_metrics.tasks import ( # noqa: E402 + AggregationTier, + aggregate_metrics_from_sources, +) + +_SPLIT_MIGRATION = "dashboard_metrics.migrations.0006_split_aggregation_schedule" +_ENDPOINT = "/internal/v1/dashboard-metrics/aggregate/" + + +def _post(body: dict[str, Any]) -> tuple[int, Any]: + """POST to the aggregate endpoint with the task mocked. + + Returns the status and the kwargs the task was called with, or ``None`` if it was + never reached — which is what a rejected body has to look like. + """ + view = internal_views.AggregateMetricsAPIView.as_view() + request = APIRequestFactory().post(_ENDPOINT, body, format="json") + with mock.patch.object( + internal_views, "aggregate_metrics_from_sources", return_value={"ok": True} + ) as task: + response = view(request) + return response.status_code, (task.call_args.kwargs if task.call_args else None) + + +class TestThePgLegCarriesTheTier: + """The endpoint the worker proxy POSTs to.""" + + @pytest.mark.parametrize("tier", ["hourly", "daily_monthly", "all"]) + def test_the_endpoint_forwards_the_tier_to_the_task(self, tier: str) -> None: + status, called_with = _post({"tier": tier}) + assert status == 200 + assert called_with == {"tier": tier} + + def test_an_omitted_tier_leaves_the_task_default_in_place(self) -> None: + """Not 'hourly', and not nothing: the task's own default is `all`, and passing + anything here would override it during the window before 0006 applies. + """ + status, called_with = _post({}) + assert status == 200 + assert called_with == {} + + @pytest.mark.parametrize("body", [{}, {"tier": None}, "not-a-dict"]) + def test_an_absent_tier_leaves_the_task_default_in_place(self, body) -> None: + """An explicit null and a non-dict body both mean "omitted", not "no tiers".""" + status, called_with = _post(body) + assert status == 200 + assert called_with == {} + + def test_the_source_window_reaches_the_task(self) -> None: + """0005's reconciliation row dispatches this against the same task path.""" + status, called_with = _post({"source_window_days": 7}) + assert status == 200 + assert called_with == {"source_window_days": 7} + + def test_both_kwargs_survive_together(self) -> None: + status, called_with = _post({"tier": "hourly", "source_window_days": 7}) + assert status == 200 + assert called_with == {"tier": "hourly", "source_window_days": 7} + + def test_a_non_integer_window_is_rejected(self) -> None: + status, called_with = _post({"source_window_days": "seven"}) + assert status == 400 + assert called_with is None + + def test_an_unrecognised_tier_is_rejected_rather_than_ignored(self) -> None: + """A silent no-op would look like a successful run that wrote nothing. + + The 400 is raised at the boundary, before the task is entered, so it cannot be + confused with a ValueError from inside the aggregation — that one belongs on + the logged 500 path. + """ + status, called_with = _post({"tier": "houry"}) + assert status == 400 + assert called_with is None + + +class TestTheBeatLegCarriesTheTier: + """Beat passes the row's stored JSON kwargs straight into the task signature.""" + + @pytest.fixture(scope="class") + def declared_kwargs(self) -> dict[str, dict[str, Any]]: + mod = importlib.import_module(_SPLIT_MIGRATION) + return {s["name"]: {"tier": s["tier"]} for s in mod.AGGREGATION_SCHEDULES} + + def test_both_rows_declare_a_tier( + self, declared_kwargs: dict[str, dict[str, Any]] + ) -> None: + assert len(declared_kwargs) == 2 + assert all("tier" in kw for kw in declared_kwargs.values()) + + def test_every_declared_kwarg_set_binds_to_the_task_signature( + self, declared_kwargs: dict[str, dict[str, Any]] + ) -> None: + """A row declaring a kwarg the task does not accept fails at call time, inside + the worker, where it surfaces as a retrying task rather than a bad schedule. + """ + signature = inspect.signature(aggregate_metrics_from_sources) + for kwargs in declared_kwargs.values(): + signature.bind(**kwargs) + + def test_every_declared_tier_is_a_real_tier( + self, declared_kwargs: dict[str, dict[str, Any]] + ) -> None: + """The migration cannot import the enum, so it repeats the literals. A typo + there raises inside the task on every single run.""" + for kwargs in declared_kwargs.values(): + AggregationTier(kwargs["tier"]) diff --git a/backend/dashboard_metrics/tests/test_aggregation_tier.py b/backend/dashboard_metrics/tests/test_aggregation_tier.py new file mode 100644 index 0000000000..f868e1395d --- /dev/null +++ b/backend/dashboard_metrics/tests/test_aggregation_tier.py @@ -0,0 +1,183 @@ +"""Guard: the tier a schedule row asks for is the tier that gets written. + +The split runs one task on two schedules that differ only in their ``tier`` kwarg, so +the gating predicates and the per-tier lock key are the whole mechanism. Each property +here is one way the split fails silently — writing nothing, writing both tiers from one +schedule, or the two schedules starving each other on the lock. + +DB-free, so this runs in the unit tier alongside test_pg_periodic_task_declarations.py. +""" + +from __future__ import annotations + +import inspect +import os +import time + +import django +import pytest +from django.apps import apps + +os.environ.setdefault("DJANGO_SETTINGS_MODULE", "backend.settings.test") +if not apps.ready: + django.setup() + +from django.core.cache import cache # noqa: E402 + +from dashboard_metrics.tasks import ( # noqa: E402 + AGGREGATION_LOCK_TIMEOUT, + DASHBOARD_SOURCE_WINDOW_DAYS, + AggregationTier, + _acquire_aggregation_lock, + _acquire_aggregation_locks, + _aggregation_lock_keys, + _tiers_written, + _writes_daily_monthly, + _writes_hourly, + aggregate_metrics_from_sources, +) + + +def _keys(tier, window: int = DASHBOARD_SOURCE_WINDOW_DAYS) -> list[str]: + return _aggregation_lock_keys(tier, window) + + +class TestWhichTiersEachRunWrites: + @pytest.mark.parametrize( + "tier,hourly,daily_monthly", + [ + (AggregationTier.HOURLY, True, False), + (AggregationTier.DAILY_MONTHLY, False, True), + (AggregationTier.ALL, True, True), + ], + ) + def test_the_predicates_partition_the_work( + self, tier: AggregationTier, hourly: bool, daily_monthly: bool + ) -> None: + assert _writes_hourly(tier) is hourly + assert _writes_daily_monthly(tier) is daily_monthly + + def test_the_two_schedules_together_cover_every_tier(self) -> None: + """Neither schedule may leave a tier unwritten: hourly and daily_monthly are + the only two rows, so between them they have to do everything `all` does. + """ + scheduled = (AggregationTier.HOURLY, AggregationTier.DAILY_MONTHLY) + assert any(_writes_hourly(t) for t in scheduled) + assert any(_writes_daily_monthly(t) for t in scheduled) + + def test_no_tier_is_written_by_both_schedules(self) -> None: + """Overlap would mean duplicate work every hour on the hour. The upserts make + it harmless, not free. + """ + assert not _writes_daily_monthly(AggregationTier.HOURLY) + assert not _writes_hourly(AggregationTier.DAILY_MONTHLY) + + +class TestTheDefaultIsAll: + """The property that keeps the deploy window safe, pinned at the signature. + + Between the code deploying and migration 0006 running, the schedule row still + carries no tier kwarg. Every other test in the suite passes a tier explicitly or + mocks the task, so none of them can see what the default actually is. + """ + + def test_the_signature_default_is_all(self) -> None: + """Narrower and daily/monthly stop being written for the whole window; none + and nothing is written at all. Both look like successful runs. + """ + default = inspect.signature(aggregate_metrics_from_sources).parameters[ + "tier" + ].default + assert default == AggregationTier.ALL + + def test_the_default_writes_everything_rather_than_nothing(self) -> None: + assert AggregationTier("all") is AggregationTier.ALL + assert _writes_hourly(AggregationTier.ALL) + assert _writes_daily_monthly(AggregationTier.ALL) + + def test_an_unrecognised_tier_raises(self) -> None: + """The internal view turns this into a 400. A silent no-op would look like a + successful run that wrote nothing. + """ + with pytest.raises(ValueError): + AggregationTier("houry") + + +class TestTheTierTableIsExhaustive: + """A member with no entry must raise, not write nothing and report success.""" + + def test_every_declared_tier_has_an_entry(self) -> None: + for tier in AggregationTier: + assert _tiers_written(tier) + + def test_an_unhandled_member_raises_rather_than_writing_nothing(self) -> None: + # Stands in for a member added to the enum without a _TIER_WRITES entry. + ghost = type("_Ghost", (), {"value": "weekly"})() + with pytest.raises(AssertionError, match="Unhandled AggregationTier"): + _tiers_written(ghost) + + +class TestTheLockCoversWhatIsWritten: + """Keyed by granularity written, not by enum member. + + Keying on the label alone gives ALL a third key that excludes nothing, so an ALL + run and the scheduled hourly run write EventMetricsHourly concurrently. These + exercise the lock rather than its key string: a version of + _acquire_aggregation_lock that ignored its argument would pass a key-shape test. + """ + + @pytest.fixture(autouse=True) + def _clear(self): + cache.clear() + yield + cache.clear() + + def test_the_two_scheduled_tiers_never_block_each_other(self) -> None: + assert _acquire_aggregation_locks(_keys(AggregationTier.HOURLY)) + assert _acquire_aggregation_locks(_keys(AggregationTier.DAILY_MONTHLY)) + + def test_a_tier_blocks_itself(self) -> None: + assert _acquire_aggregation_locks(_keys(AggregationTier.HOURLY)) + assert not _acquire_aggregation_locks(_keys(AggregationTier.HOURLY)) + + def test_all_is_blocked_by_either_half(self) -> None: + """The exclusion a per-member key silently dropped.""" + assert _acquire_aggregation_locks(_keys(AggregationTier.HOURLY)) + assert not _acquire_aggregation_locks(_keys(AggregationTier.ALL)) + + def test_a_blocked_run_releases_whatever_it_took(self) -> None: + """ALL takes hourly first; failing on daily_monthly must not strand hourly.""" + assert _acquire_aggregation_locks(_keys(AggregationTier.DAILY_MONTHLY)) + assert not _acquire_aggregation_locks(_keys(AggregationTier.ALL)) + assert _acquire_aggregation_locks(_keys(AggregationTier.HOURLY)) + + def test_a_wider_window_is_a_different_job(self) -> None: + """The reconciliation pass is never retried, so it must not be starved by the + 15-minute schedule it races against.""" + assert _acquire_aggregation_locks(_keys(AggregationTier.HOURLY, 2)) + assert _acquire_aggregation_locks(_keys(AggregationTier.ALL, 7)) + + +class TestTheLockSelfHeals: + """Both reclaim branches, neither of which was executed by any test.""" + + @pytest.fixture(autouse=True) + def _clear(self): + cache.clear() + yield + cache.clear() + + def test_a_lock_older_than_the_timeout_is_reclaimed(self) -> None: + key = _keys(AggregationTier.HOURLY)[0] + cache.set(key, str(time.time() - AGGREGATION_LOCK_TIMEOUT - 1), 3600) + assert _acquire_aggregation_lock(key) + + def test_a_fresh_lock_is_not_reclaimed(self) -> None: + key = _keys(AggregationTier.HOURLY)[0] + cache.set(key, str(time.time()), 3600) + assert not _acquire_aggregation_lock(key) + + def test_a_corrupted_lock_value_is_reclaimed(self) -> None: + key = _keys(AggregationTier.HOURLY)[0] + cache.set(key, "running", 3600) + assert _acquire_aggregation_lock(key) diff --git a/backend/dashboard_metrics/tests/test_migration_graph.py b/backend/dashboard_metrics/tests/test_migration_graph.py new file mode 100644 index 0000000000..d5cfbbbf55 --- /dev/null +++ b/backend/dashboard_metrics/tests/test_migration_graph.py @@ -0,0 +1,65 @@ +"""Guard: the migration graph builds (UN-3974). + +Django builds the **entire** graph before executing anything, so one migration +depending on a node that does not exist aborts `migrate`, `makemigrations` and +`showmigrations` for every app in the project — the deploy's migrate step fails, not +just this app's. + +Nothing else catches it. The backend suite runs with `--no-migrations`, so test-DB +creation never builds the graph, and every migration test in this app reaches its +module through `importlib.import_module`, which resolves a file path rather than a +graph node. GitHub also reports a stacked branch as mergeable, because a missing +dependency is not a textual conflict. + +DB-free: building the graph reads the migration files, not the database. +""" + +from __future__ import annotations + +import os + +import django +from django.apps import apps + +os.environ.setdefault("DJANGO_SETTINGS_MODULE", "backend.settings.test") +if not apps.ready: + django.setup() + +from django.db.migrations.loader import MigrationLoader # noqa: E402 +from django.test import SimpleTestCase, override_settings # noqa: E402 + + +def _build_graph() -> MigrationLoader: + """Build the real graph, whatever the suite's own flags say. + + `--no-migrations` works by pointing MIGRATION_MODULES at a mapping that returns + None for every app, so a loader built under it finds nothing and every assertion + below would pass against an empty graph. Restoring the setting is what makes this + guard mean anything in the tier it runs in. + """ + with override_settings(MIGRATION_MODULES={}): + loader = MigrationLoader(None, ignore_no_migrations=True) + loader.build_graph() + return loader + + +class MigrationGraphTests(SimpleTestCase): + def test_the_graph_builds(self) -> None: + """A dependency on an absent migration raises NodeNotFoundError here.""" + loader = _build_graph() + self.assertTrue(loader.graph.nodes, "no migrations loaded — the guard is inert") + + def test_every_app_has_exactly_one_leaf(self) -> None: + """Two leaves in one app block `migrate` for every app, not just that one. + + This is what a merge of two branches that each added a migration produces, and + it is invisible until deploy for the same `--no-migrations` reason. + """ + loader = _build_graph() + + leaves: dict[str, list[str]] = {} + for app_label, name in loader.graph.leaf_nodes(): + leaves.setdefault(app_label, []).append(name) + + conflicts = {app: names for app, names in leaves.items() if len(names) > 1} + self.assertEqual(conflicts, {}, f"apps with multiple leaf migrations: {conflicts}") diff --git a/backend/dashboard_metrics/tests/test_pg_periodic_task_declarations.py b/backend/dashboard_metrics/tests/test_pg_periodic_task_declarations.py index d83e2c2c69..b815aafa07 100644 --- a/backend/dashboard_metrics/tests/test_pg_periodic_task_declarations.py +++ b/backend/dashboard_metrics/tests/test_pg_periodic_task_declarations.py @@ -13,6 +13,11 @@ Migrations are run in order against fake models, so rows a later migration rewrites are compared in their final state. +``0006_split_aggregation_schedule`` then splits the aggregation into two rows by tier. +It writes both scheduler tables from one spec, so the new row cannot drift by +construction — but it also rewrites an existing row, and *how* it does that is +load-bearing. The last sections cover that, the ownership it inherits, and the rollback. + DB-free — nothing here touches a database. """ @@ -21,14 +26,24 @@ import importlib import inspect import json +import os import re from pathlib import Path from types import SimpleNamespace +from typing import Any +import django import pytest -from django.db import migrations +from django.apps import apps + +os.environ.setdefault("DJANGO_SETTINGS_MODULE", "backend.settings.test") +if not apps.ready: + django.setup() -from dashboard_metrics.tasks import ( +from django.db import migrations # noqa: E402 + +from dashboard_metrics.tasks import ( # noqa: E402 + AggregationTier, aggregate_metrics_from_sources, cleanup_daily_metrics, cleanup_hourly_metrics, @@ -44,6 +59,8 @@ "dashboard_metrics_cleanup_hourly": "0 2 * * *", "dashboard_metrics_cleanup_daily": "0 3 * * 0", "dashboard_metrics_reconcile_source_window": "40 4 * * *", + # Added by 0006; off the */15 grid so it never starts alongside the hourly tier. + "dashboard_metrics_aggregate_daily_monthly": "20 * * * *", } @@ -93,6 +110,13 @@ def filter(self, name=None, name__in=None, **_kwargs): self._selected = [name] if name is not None else list(name__in or []) return self + def first(self): + """The row a migration reads back, e.g. to inherit scheduler ownership.""" + name = self._selected[0] if self._selected else None + if name not in self.rows: + return None + return SimpleNamespace(**{"enabled": True, "pg_owned": False, **self.rows[name]}) + def update(self, **kwargs): self.writes += 1 for name in self._selected: @@ -258,3 +282,280 @@ def test_no_row_presets_a_run_time(self, declared): for row in declared.pg.values(): assert "next_run_at" not in row assert "last_run_at" not in row + + +_SPLIT_MIGRATION = "dashboard_metrics.migrations.0006_split_aggregation_schedule" +_NEW_ROW = "dashboard_metrics_aggregate_daily_monthly" +_EXISTING_ROW = "dashboard_metrics_aggregate_from_sources" + + +class _SplitRecorder: + """Captures what 0006 does to one scheduler table, keeping creates and updates apart. + + The distinction is the point: creating a row writes every default, updating one writes + only the named fields. Conflating them is exactly the bug this guards. + """ + + def __init__(self, existing: Any = None) -> None: + self.created: dict[str, dict[str, Any]] = {} + self.updated: dict[str, dict[str, Any]] = {} + self.deleted: list[str] = [] + self.bumps = 0 + # How many rows a filtered update matches. 0 models the row being absent, + # which is the case the migration now refuses to report as success. + self.rows_present: int | None = None + self._existing = existing + self._filtered_on: str = "" + self._filtered_in: list[str] = [] + + def filter( + self, name: str = "", name__in: list[str] | None = None, **_kw: Any + ) -> _SplitRecorder: + self._filtered_on = name + self._filtered_in = list(name__in or []) + return self + + def first(self) -> Any: + return self._existing + + def update(self, **kwargs: Any) -> int: + self.updated[self._filtered_on] = kwargs + return 1 if self.rows_present is None else self.rows_present + + def update_or_create( + self, name: str = "", defaults: dict[str, Any] | None = None, **_kw: Any + ) -> tuple[dict[str, Any], bool]: + if not name: # PeriodicTasks(ident=1) — the Beat reload tracker + self.bumps += 1 + return defaults or {}, True + self.created[name] = defaults or {} + return self.created[name], True + + def get_or_create(self, **kwargs: Any) -> tuple[dict[str, Any], bool]: + return kwargs, True + + def delete(self) -> tuple[int, dict[str, Any]]: + self.deleted.extend(self._filtered_in or [self._filtered_on]) + return (len(self.deleted), {}) + + +def _run_split(beat_row: Any = None, pg_row: Any = None) -> dict[str, _SplitRecorder]: + """Run 0006's forward function against fakes and capture every table it writes.""" + mod = importlib.import_module(_SPLIT_MIGRATION) + beat = _SplitRecorder(existing=beat_row) + pg = _SplitRecorder(existing=pg_row) + crontab, tracker = _SplitRecorder(), _SplitRecorder() + + class _Apps: + def get_model(self, _app: str, model: str) -> type: + table = { + "PeriodicTask": beat, + "PgPeriodicTask": pg, + "PeriodicTasks": tracker, + }.get(model, crontab) + return type("M", (), {"objects": table}) + + mod.split_schedules(_Apps(), None) + return {"beat": beat, "pg": pg, "tracker": tracker} + + +@pytest.fixture(scope="module") +def split() -> dict[str, _SplitRecorder]: + """The default case: a Beat-owned row, as every environment ships today.""" + return _run_split( + beat_row=SimpleNamespace(enabled=True), + pg_row=SimpleNamespace(enabled=True, pg_owned=False), + ) + + +class TestTheSplitAddsOneRowAndRewritesOne: + def test_only_the_daily_monthly_row_is_created(self, split: dict[str, _SplitRecorder]) -> None: + for table in ("beat", "pg"): + assert set(split[table].created) == {_NEW_ROW} + + def test_only_the_existing_aggregate_row_is_updated(self, split: dict[str, _SplitRecorder]) -> None: + for table in ("beat", "pg"): + assert set(split[table].updated) == {_EXISTING_ROW} + + def test_the_new_row_is_declared_the_same_on_both_tables(self, split: dict[str, _SplitRecorder]) -> None: + beat, pg = split["beat"].created[_NEW_ROW], split["pg"].created[_NEW_ROW] + assert pg["task_name"] == beat["task"] + assert pg["queue"] == beat["queue"] + assert pg["task_kwargs"] == json.loads(beat["kwargs"]) + + def test_the_new_row_runs_hourly_on_both_tables(self, split: dict[str, _SplitRecorder]) -> None: + assert split["pg"].created[_NEW_ROW]["cron_string"] == "20 * * * *" + crontab = split["beat"].created[_NEW_ROW]["crontab"] + assert (crontab["minute"], crontab["hour"]) == ("20", "*") + + def test_the_two_rows_never_start_together(self, split: dict[str, _SplitRecorder]) -> None: + """The per-tier locks are built so the two runs cannot block each other, so a + shared minute is two full prefilter scans and two per-org loops at once — on a + change whose object is flattening cron load. + """ + fires_at = {0, 15, 30, 45} # the existing row's */15 + minute = int(split["beat"].created[_NEW_ROW]["crontab"]["minute"]) + assert minute not in fires_at + + def test_the_new_row_is_seeded_inert_on_the_pg_side(self, split: dict[str, _SplitRecorder]) -> None: + """Same reason as 0004's rows: a PG row that is pg_owned before the scheduler + has adopted it would fire alongside its Beat twin. + """ + assert split["pg"].created[_NEW_ROW]["pg_owned"] is False + + def test_the_rewritten_row_carries_the_same_kwargs_on_both_tables( + self, split: dict[str, _SplitRecorder] + ) -> None: + """The row firing the hourly tier every 15 minutes in production. + + Beat's ``kwargs`` is a TextField it parses with ``json.loads``; writing the + mapping rather than its JSON encoding stores a Python repr, ``ModelEntry`` + raises, and the hourly aggregation silently stops firing. + """ + beat = split["beat"].updated[_EXISTING_ROW] + assert json.loads(beat["kwargs"]) == split["pg"].updated[_EXISTING_ROW][ + "task_kwargs" + ] + + def test_the_two_rows_ask_for_different_tiers(self, split: dict[str, _SplitRecorder]) -> None: + new = split["pg"].created[_NEW_ROW]["task_kwargs"]["tier"] + existing = split["pg"].updated[_EXISTING_ROW]["task_kwargs"]["tier"] + assert new != existing + + +class TestTheNewRowInheritsWhoeverFiresTheRowItSplitsFrom: + """Hardcoding Beat leaves the daily/monthly tier with no firer in a PG-adopted + environment: the adopted row's Beat twin is disabled and Beat may be scaled to + zero, so the sole writer of those figures never runs and the hourly run still + reports success. + """ + + def test_a_pg_adopted_row_hands_its_new_half_to_the_pg_scheduler(self) -> None: + split = _run_split( + beat_row=SimpleNamespace(enabled=False), + pg_row=SimpleNamespace(enabled=True, pg_owned=True), + ) + assert split["pg"].created[_NEW_ROW]["pg_owned"] is True + assert split["pg"].created[_NEW_ROW]["enabled"] is True + assert split["beat"].created[_NEW_ROW]["enabled"] is False + + def test_a_disabled_row_does_not_come_back_as_an_enabled_half(self) -> None: + split = _run_split( + beat_row=SimpleNamespace(enabled=False), + pg_row=SimpleNamespace(enabled=False, pg_owned=False), + ) + assert split["beat"].created[_NEW_ROW]["enabled"] is False + assert split["pg"].created[_NEW_ROW]["enabled"] is False + + def test_a_missing_row_falls_back_to_beat(self) -> None: + """A fresh install applies 0002/0004 first, so this is defensive only.""" + split = _run_split(beat_row=None, pg_row=None) + assert split["beat"].created[_NEW_ROW]["enabled"] is True + assert split["pg"].created[_NEW_ROW]["pg_owned"] is False + + +class TestARunningBeatIsToldToReload: + """Historical models fire no post_save, so DatabaseScheduler never reloads. + + Without an explicit bump the existing row keeps firing with no tier and the new + row never fires at all — no error, nothing logged, and the whole saving silently + does not happen. + """ + + def test_the_forward_direction_bumps_the_change_tracker(self, split) -> None: + assert split["tracker"].bumps == 1 + + def test_the_reverse_direction_bumps_it_too(self) -> None: + mod = importlib.import_module(_SPLIT_MIGRATION) + tracker, other = _SplitRecorder(), _SplitRecorder() + + class _Apps: + def get_model(self, _app: str, model: str) -> type: + table = tracker if model == "PeriodicTasks" else other + return type("M", (), {"objects": table}) + + mod.merge_schedules(_Apps(), None) + assert tracker.bumps == 1 + + +class TestTheFrozenLiteralsMatchTheEnumToday: + """These are wire values, so a rename has to fail loudly rather than pass. + + The migration cannot import the enum, and it never re-runs — so renaming an + AggregationTier value and "keeping this in step" leaves live rows carrying the old + string while every test goes green. Comparing the two here turns that into a + failure at the moment of the rename. + """ + + def test_the_declared_tiers_are_exactly_the_schedulable_ones(self) -> None: + mod = importlib.import_module(_SPLIT_MIGRATION) + declared = {spec["tier"] for spec in mod.AGGREGATION_SCHEDULES} + # ALL is the signature default and the pre-migration row's meaning; no + # schedule row ever carries it. + schedulable = {t.value for t in AggregationTier} - {AggregationTier.ALL.value} + assert declared == schedulable + + +class TestTheRollbackRestoresOneRow: + """merge_schedules is this PR's stated safety story and had no coverage at all.""" + + def _run_merge(self, rows_present: int | None = None): + mod = importlib.import_module(_SPLIT_MIGRATION) + beat, pg, tracker = _SplitRecorder(), _SplitRecorder(), _SplitRecorder() + beat.rows_present = rows_present + pg.rows_present = rows_present + + class _Apps: + def get_model(self, _app: str, model: str) -> type: + table = { + "PeriodicTask": beat, + "PgPeriodicTask": pg, + "PeriodicTasks": tracker, + }.get(model, _SplitRecorder()) + return type("M", (), {"objects": table}) + + mod.merge_schedules(_Apps(), None) + return {"beat": beat, "pg": pg, "tracker": tracker} + + def test_the_added_row_is_deleted_from_both_tables(self) -> None: + merged = self._run_merge() + for table in ("beat", "pg"): + assert _NEW_ROW in merged[table].deleted + + def test_the_existing_row_gets_its_pre_split_payload_back(self) -> None: + merged = self._run_merge() + assert merged["beat"].updated[_EXISTING_ROW]["kwargs"] == "{}" + assert merged["pg"].updated[_EXISTING_ROW]["task_kwargs"] == {} + + def test_the_rollback_leaves_ownership_alone_like_the_forward_direction(self) -> None: + merged = self._run_merge() + assert "enabled" not in merged["beat"].updated[_EXISTING_ROW] + assert set(merged["pg"].updated[_EXISTING_ROW]) == {"task_kwargs"} + + def test_a_rollback_that_restores_nothing_raises(self) -> None: + """A bulk update matching no row would otherwise report a clean rollback while + leaving the daily and monthly tiers with no schedule at all.""" + with pytest.raises(RuntimeError, match=_EXISTING_ROW): + self._run_merge(rows_present=0) + + +class TestTheRewriteLeavesSchedulerOwnershipAlone: + """The existing row may already be owned by the PG scheduler, with its Beat twin + disabled by converge_pg_scheduler. Rewriting `pg_owned` or `enabled` here would + hand it back — and since the Beat twin stays disabled, the aggregation would be + left with no firer at all. Only the payload may change. + """ + + def test_the_pg_update_touches_only_the_kwargs(self, split: dict[str, _SplitRecorder]) -> None: + assert set(split["pg"].updated[_EXISTING_ROW]) == {"task_kwargs"} + + def test_the_beat_update_does_not_re_enable_the_row(self, split: dict[str, _SplitRecorder]) -> None: + assert "enabled" not in split["beat"].updated[_EXISTING_ROW] + + def test_the_existing_row_keeps_its_cadence(self, split: dict[str, _SplitRecorder]) -> None: + """Only the daily/monthly half moves to hourly; the hourly tier stays at 15 + minutes, which is the first half of the ticket's acceptance criteria. + """ + for table in ("beat", "pg"): + update = split[table].updated[_EXISTING_ROW] + assert not {"crontab", "interval", "cron_string"} & set(update) diff --git a/backend/dashboard_metrics/tests/test_tasks.py b/backend/dashboard_metrics/tests/test_tasks.py index e7911226d6..2e7a349e10 100644 --- a/backend/dashboard_metrics/tests/test_tasks.py +++ b/backend/dashboard_metrics/tests/test_tasks.py @@ -35,9 +35,11 @@ AGGREGATION_LOCK_TIMEOUT, DASHBOARD_RECONCILE_WINDOW_DAYS, DASHBOARD_SOURCE_WINDOW_DAYS, + AggregationTier, _acquire_aggregation_lock, + _acquire_aggregation_locks, _active_org_ids, - _aggregation_lock_key, + _aggregation_lock_keys, _rollup_monthly_from_daily, _run_aggregation, _truncate_to_day, @@ -764,41 +766,40 @@ def test_the_comparison_can_fail_when_the_daily_tier_is_wrong(self): class TestTheLockIsPerSchedule(TestCase): - """The reconciliation pass must not lose a race it is never retried after.""" + """The reconciliation pass must not lose a race it is never retried after. - def test_the_two_schedules_take_different_keys(self): - assert _aggregation_lock_key( - DASHBOARD_SOURCE_WINDOW_DAYS - ) != _aggregation_lock_key(DASHBOARD_RECONCILE_WINDOW_DAYS) + Per-granularity exclusion is covered in test_aggregation_tier.py; this is the + window half — two schedules that both write every tier. + """ - def test_a_held_key_does_not_block_the_other_schedule(self): + def setUp(self): cache.clear() - assert _acquire_aggregation_lock( - _aggregation_lock_key(DASHBOARD_SOURCE_WINDOW_DAYS) + self.addCleanup(cache.clear) + + def _keys(self, window): + return _aggregation_lock_keys(AggregationTier.ALL, window) + + def test_the_two_schedules_take_different_keys(self): + assert self._keys(DASHBOARD_SOURCE_WINDOW_DAYS) != self._keys( + DASHBOARD_RECONCILE_WINDOW_DAYS ) + + def test_a_held_key_does_not_block_the_other_schedule(self): + assert _acquire_aggregation_locks(self._keys(DASHBOARD_SOURCE_WINDOW_DAYS)) # Same schedule: excluded, which is what the lock is for. - assert not _acquire_aggregation_lock( - _aggregation_lock_key(DASHBOARD_SOURCE_WINDOW_DAYS) - ) + assert not _acquire_aggregation_locks(self._keys(DASHBOARD_SOURCE_WINDOW_DAYS)) # The reconciliation pass proceeds regardless. - assert _acquire_aggregation_lock( - _aggregation_lock_key(DASHBOARD_RECONCILE_WINDOW_DAYS) - ) - cache.clear() + assert _acquire_aggregation_locks(self._keys(DASHBOARD_RECONCILE_WINDOW_DAYS)) def test_a_stale_lock_is_reclaimed(self): - cache.clear() - key = _aggregation_lock_key(DASHBOARD_SOURCE_WINDOW_DAYS) + key = self._keys(DASHBOARD_SOURCE_WINDOW_DAYS)[0] cache.set(key, str(time.time() - AGGREGATION_LOCK_TIMEOUT - 1), 3600) assert _acquire_aggregation_lock(key) - cache.clear() def test_a_corrupted_lock_value_is_reclaimed(self): - cache.clear() - key = _aggregation_lock_key(DASHBOARD_SOURCE_WINDOW_DAYS) + key = self._keys(DASHBOARD_SOURCE_WINDOW_DAYS)[0] cache.set(key, "running", 3600) assert _acquire_aggregation_lock(key) - cache.clear() class TestSourceWindowValidation(TestCase): @@ -872,7 +873,9 @@ def test_task_passes_the_window_through(self): patch("dashboard_metrics.tasks._run_aggregation") as mock_run, ): aggregate_metrics_from_sources() - mock_run.assert_called_once_with(DASHBOARD_SOURCE_WINDOW_DAYS) + mock_run.assert_called_once_with( + AggregationTier.ALL, DASHBOARD_SOURCE_WINDOW_DAYS + ) with ( patch("dashboard_metrics.tasks._acquire_aggregation_lock", return_value=True), @@ -880,7 +883,7 @@ def test_task_passes_the_window_through(self): patch("dashboard_metrics.tasks._run_aggregation") as mock_run, ): aggregate_metrics_from_sources(source_window_days=7) - mock_run.assert_called_once_with(7) + mock_run.assert_called_once_with(AggregationTier.ALL, 7) def _seed_file( self, days_ago: int, status: ExecutionStatus = ExecutionStatus.COMPLETED diff --git a/backend/dashboard_metrics/tests/test_tier_split_equivalence.py b/backend/dashboard_metrics/tests/test_tier_split_equivalence.py new file mode 100644 index 0000000000..3f6fbe54b2 --- /dev/null +++ b/backend/dashboard_metrics/tests/test_tier_split_equivalence.py @@ -0,0 +1,228 @@ +"""The split preserves every figure it used to write (UN-3974, AC-2). + +AC-2 is an equivalence claim — "hourly figures unchanged; daily and monthly lag by at +most one hour" — so it is settled by running the real aggregation and diffing what lands +in the metrics tables, not by reasoning about the gating predicates. Those are pinned +separately in test_aggregation_tier.py; this is the outcome they are supposed to produce. + +Two properties, and both matter: + +- the `hourly` schedule reproduces what an all-tiers run writes to EventMetricsHourly, + exactly — that is the "unchanged" half. Note this is a **partition** property of the + post-change code, not a comparison against the pre-split implementation, which this + branch does not have: `test_the_hourly_tier_holds_the_figures_the_fixture_implies` + is what anchors it to an absolute number +- `hourly` and `daily_monthly` together reproduce every row the pre-split run wrote to + any table — that is the "nothing is lost" half, which the AC assumes rather than states + +DB-bound, so conftest marks it integration. +""" + +from __future__ import annotations + +import uuid +from datetime import timedelta +from typing import Any +from unittest.mock import patch + +import os + +import django +from django.apps import apps + +os.environ.setdefault("DJANGO_SETTINGS_MODULE", "backend.settings.test") +if not apps.ready: + django.setup() + +from account_v2.models import Organization # noqa: E402 +from django.db import connection # noqa: E402 +from django.test import TestCase # noqa: E402 +from django.utils import timezone # noqa: E402 +from workflow_manager.workflow_v2.models.workflow import Workflow # noqa: E402 + +from dashboard_metrics.models import ( # noqa: E402 + EventMetricsDaily, + EventMetricsHourly, + EventMetricsMonthly, +) +from dashboard_metrics.tasks import ( # noqa: E402 + AggregationTier, + _run_aggregation, + _truncate_to_month, +) + +# (model, the column naming its period) — the period field differs per tier. +_TIERS = [ + (EventMetricsHourly, "timestamp"), + (EventMetricsDaily, "date"), + (EventMetricsMonthly, "month"), +] +_FIELDS = ["metric_name", "metric_type", "metric_value", "metric_count"] + + +class TestTheSplitPreservesEveryFigure(TestCase): + def setUp(self) -> None: + self.org = Organization.objects.create( + organization_id="tier-split-org", name="tier-split", display_name="Tier Split" + ) + self.workflow = Workflow.objects.create( + workflow_name="tier-split-wf", organization=self.org + ) + self.now = now = timezone.now() + # One row per window the aggregation reads — last 24h for the hourly tier, last + # 7 days for daily, inside the previous month for monthly. The two recent ones + # also make the org visible to the active-org prefilter, without which nothing + # runs at all. + # + # The previous-month row is derived from the month boundary, not a fixed + # "25 days ago": for the last few days of any month that lands in the *current* + # month and the cross-boundary coverage silently disappears. + last_month_day = _truncate_to_month(now) - timedelta(days=1) + windows = [ + now - timedelta(hours=2), + now - timedelta(hours=5), + now - timedelta(days=3), + last_month_day, + ] + executions = self._add_executions(windows) + # Both aggregation paths have to be exercised: the per-metric queries go through + # _aggregate_single_metric and the four LLM metrics through + # _aggregate_llm_combined, and each gates on the tier separately. A fixture + # producing only LLM figures leaves half the split unverified. + self._add_file_executions(executions) + self._add_llm_usage(windows) + + def _add_executions(self, timestamps: list[Any]) -> list[tuple[Any, Any]]: + """Raw insert so created_at is ours; the model sets it with auto_now_add.""" + created = [] + with connection.cursor() as cur: + for ts in timestamps: + execution_id = uuid.uuid4() + cur.execute( + "INSERT INTO workflow_execution (id, created_at, modified_at, " + "workflow_id, execution_mode, execution_method, execution_type, " + "execution_log_id, status, error_message, attempts, execution_time, " + "result_acknowledged, total_files) " + "VALUES (%s, %s, %s, %s, 'INSTANT', 'DIRECT', 'COMPLETE', '', " + "'COMPLETED', '', 0, 1.0, false, 1)", + [execution_id, ts, ts, self.workflow.id], + ) + created.append((execution_id, ts)) + return created + + def _add_file_executions(self, executions: list[tuple[Any, Any]]) -> None: + """Feeds documents_processed, which runs through _aggregate_single_metric.""" + with connection.cursor() as cur: + for execution_id, ts in executions: + cur.execute( + "INSERT INTO workflow_file_execution (id, created_at, modified_at, " + "file_name, status, workflow_execution_id) " + "VALUES (%s, %s, %s, 'doc.pdf', 'COMPLETED', %s)", + [uuid.uuid4(), ts, ts, execution_id], + ) + + def _add_llm_usage(self, timestamps: list[Any]) -> None: + """LLM metrics need no joins, so they are the cheapest way to put a real figure + in all three tiers.""" + with connection.cursor() as cur: + for ts in timestamps: + cur.execute( + "INSERT INTO usage (id, created_at, modified_at, adapter_instance_id, " + "usage_type, llm_usage_reason, model_name, embedding_tokens, " + "prompt_tokens, completion_tokens, total_tokens, cost_in_dollars, " + "organization_id) " + "VALUES (%s, %s, %s, 'test-adapter', 'llm', 'extraction', 'test-model', " + "0, 100, 50, 150, 0.25, %s)", + [uuid.uuid4(), ts, ts, self.org.id], + ) + + def _snapshot(self) -> dict[str, set[tuple[Any, ...]]]: + return { + model.__name__: set( + model._base_manager.values_list("organization_id", period, *_FIELDS) + ) + for model, period in _TIERS + } + + def _clear(self) -> None: + for model, _ in _TIERS: + model._base_manager.all().delete() + + def _run(self, tier: AggregationTier) -> dict[str, set[tuple[Any, ...]]]: + """One clock for every run in a test. + + _run_aggregation reads timezone.now() itself, so three unpatched invocations + compute three different window starts — and a run straddling an hour or a month + boundary would fail on a non-regression. + """ + self._clear() + with patch("dashboard_metrics.tasks.timezone.now", return_value=self.now): + _run_aggregation(tier) + return self._snapshot() + + def test_the_pre_split_run_writes_all_three_tiers(self) -> None: + """Guards the tests below from passing vacuously: an equivalence between two + empty sets proves nothing. + """ + every_tier = self._run(AggregationTier.ALL) + for name, rows in every_tier.items(): + assert rows, f"{name} is empty — the fixture produces no metrics to compare" + + def test_the_fixture_exercises_both_aggregation_paths(self) -> None: + """The other way these tests can go quietly vacuous. The tier is checked + separately in _aggregate_single_metric and in _aggregate_llm_combined, so a + fixture yielding only one kind of metric verifies only half the split — which + is exactly what a mutation test caught here. + """ + every_tier = self._run(AggregationTier.ALL) + for table, rows in every_tier.items(): + names = {row[2] for row in rows} + assert "documents_processed" in names, f"{table}: no per-metric figure" + assert "llm_calls" in names, f"{table}: no combined-LLM figure" + + def test_the_hourly_tier_holds_the_figures_the_fixture_implies(self) -> None: + """An absolute expectation, not a comparison of the code against itself. + + Every other assertion in this file runs the same post-change function twice, so + a regression in the shared path — the window arithmetic, the org_identifier + handoff, an upsert that writes zeroes — moves both sides equally and stays + green. This one names a number the fixture determines. + """ + self._run(AggregationTier.HOURLY) + rows = EventMetricsHourly._base_manager.filter( + metric_name="documents_processed" + ) + assert sum(row.metric_value for row in rows) == 2, ( + "exactly the -2h and -5h file executions fall inside the 24h window; " + "the -3d and previous-month ones must not" + ) + assert {row.metric_value for row in rows} != {0} + + def test_hourly_reproduces_the_pre_split_hourly_figures(self) -> None: + """The "figures unchanged" half of AC-2, row for row rather than in aggregate.""" + before = self._run(AggregationTier.ALL)["EventMetricsHourly"] + after = self._run(AggregationTier.HOURLY)["EventMetricsHourly"] + assert after == before + + def test_the_two_schedules_together_lose_nothing(self) -> None: + """Every row the single pre-split run wrote is still written by one of the two + schedules, and neither invents one. + """ + every_tier = self._run(AggregationTier.ALL) + hourly = self._run(AggregationTier.HOURLY) + daily_monthly = self._run(AggregationTier.DAILY_MONTHLY) + + for name in every_tier: + combined = hourly[name] | daily_monthly[name] + assert combined == every_tier[name], f"{name} differs after the split" + + def test_neither_schedule_writes_the_other_tiers_tables(self) -> None: + """If they overlapped, the two schedules would duplicate work every hour on the + hour — harmless thanks to the upserts, but not free. + """ + hourly = self._run(AggregationTier.HOURLY) + assert not hourly["EventMetricsDaily"] + assert not hourly["EventMetricsMonthly"] + + daily_monthly = self._run(AggregationTier.DAILY_MONTHLY) + assert not daily_monthly["EventMetricsHourly"] diff --git a/backend/workflow_manager/workflow_v2/migrations/0029_we_created_at_idx.py b/backend/workflow_manager/workflow_v2/migrations/0029_we_created_at_idx.py new file mode 100644 index 0000000000..11054900fd --- /dev/null +++ b/backend/workflow_manager/workflow_v2/migrations/0029_we_created_at_idx.py @@ -0,0 +1,63 @@ +"""Add a created_at index to workflow_execution. + +Serves bare "rows in this date window" queries with no leading column value — the +dashboard metrics active-org prefilter today, and the grouped metric queries in +UN-4045. The composite indexes lead with workflow_id / pipeline_id, so they are +date-ordered only within one workflow or pipeline; the partial index is empty in +steady state. Measurements in UN-3883. + +Built CONCURRENTLY (atomic = False): a plain AddIndex holds a SHARE lock for the +whole build and would block every execution in flight. Prefer building it out of +band before the deploy; the migration then no-ops via IF NOT EXISTS. +""" + +from django.db import migrations, models + +INDEX_NAME = "we_created_at_idx" + +# An interrupted CONCURRENTLY build leaves an INVALID index that costs on every +# write and is never read. IF NOT EXISTS would keep it while Django recorded the +# migration as applied, so fail loudly instead. +_ASSERT_INDEX_VALID = f""" +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 FROM pg_class c + JOIN pg_index i ON i.indexrelid = c.oid + WHERE c.relname = '{INDEX_NAME}' AND NOT i.indisvalid + ) THEN + RAISE EXCEPTION 'Index {INDEX_NAME} exists but is INVALID (a prior CREATE INDEX CONCURRENTLY was interrupted). Drop it and re-run: DROP INDEX CONCURRENTLY IF EXISTS {INDEX_NAME};'; + END IF; +END +$$; +""" + + +class Migration(migrations.Migration): + # CREATE / DROP INDEX CONCURRENTLY cannot run inside a transaction block. + atomic = False + + dependencies = [("workflow_v2", "0028_undispatched_idx_dispatched_at")] + + operations = [ + migrations.SeparateDatabaseAndState( + database_operations=[ + migrations.RunSQL( + sql=( + f"CREATE INDEX CONCURRENTLY IF NOT EXISTS {INDEX_NAME} " + "ON workflow_execution (created_at);" + ), + reverse_sql=f"DROP INDEX CONCURRENTLY IF EXISTS {INDEX_NAME};", + ), + migrations.RunSQL( + sql=_ASSERT_INDEX_VALID, reverse_sql=migrations.RunSQL.noop + ), + ], + state_operations=[ + migrations.AddIndex( + model_name="workflowexecution", + index=models.Index(fields=["created_at"], name=INDEX_NAME), + ), + ], + ), + ] diff --git a/backend/workflow_manager/workflow_v2/models/execution.py b/backend/workflow_manager/workflow_v2/models/execution.py index d6082aa423..a8104898d9 100644 --- a/backend/workflow_manager/workflow_v2/models/execution.py +++ b/backend/workflow_manager/workflow_v2/models/execution.py @@ -272,6 +272,9 @@ class Meta: queue_message_id__isnull=True, ), ), + # Bare created_at range scans; the indexes above are date-ordered only + # within one workflow or pipeline. See migration 0029. + models.Index(fields=["created_at"], name="we_created_at_idx"), ] @property diff --git a/backend/workflow_manager/workflow_v2/tests/test_we_created_at_idx.py b/backend/workflow_manager/workflow_v2/tests/test_we_created_at_idx.py new file mode 100644 index 0000000000..f4f1d2d759 --- /dev/null +++ b/backend/workflow_manager/workflow_v2/tests/test_we_created_at_idx.py @@ -0,0 +1,125 @@ +"""Guard: ``we_created_at_idx`` keeps the shape that makes it safe to deploy. + +The backend suite runs with ``--no-migrations``, so migration 0029 never executes in +CI. Regenerating it with ``makemigrations``, or dropping ``atomic = False`` while +tidying, lands a plain ``AddIndex`` — which holds a SHARE lock for the whole build and +blocks every in-flight execution on a multi-million-row table — with every other test +still green. These assert the properties that keep that from happening. + +Model and migration introspection only, no test database, so this runs in the unit tier +alongside ``test_active_execution_index.py`` and ``test_undispatched_execution_index.py``. +""" + +from __future__ import annotations + +import importlib +import os +import re +from pathlib import Path +from typing import Any, cast + +import django +from django.apps import apps +from django.db import migrations, models + +os.environ.setdefault("DJANGO_SETTINGS_MODULE", "backend.settings.test") +if not apps.ready: + django.setup() + +INDEX_NAME = "we_created_at_idx" +_MIGRATION_FILE = ( + Path(__file__).resolve().parent.parent / "migrations" / "0029_we_created_at_idx.py" +) +_MIGRATION_MODULE = "workflow_manager.workflow_v2.migrations.0029_we_created_at_idx" + + +def _model_index() -> models.Index | None: + model = apps.get_model("workflow_v2", "WorkflowExecution") + return next((i for i in model._meta.indexes if i.name == INDEX_NAME), None) + + +def _operations() -> list[Any]: + return cast( + list[Any], importlib.import_module(_MIGRATION_MODULE).Migration.operations + ) + + +class TestTheModelDeclaresIt: + def test_it_is_keyed_on_created_at_alone(self) -> None: + """A bare created_at range with no leading column value is the whole point — + the composite indexes lead with workflow_id / pipeline_id and are date-ordered + only within one workflow or pipeline. + """ + index = _model_index() + assert index is not None, f"{INDEX_NAME} is missing from WorkflowExecution.Meta" + assert index.fields == ["created_at"] + + def test_it_carries_no_condition(self) -> None: + """A partial index would not serve the prefilter, which bounds nothing but the + date. we_undispatched_dispatch_idx is the partial one and is a different index. + """ + assert getattr(_model_index(), "condition", None) is None + + +class TestTheMigrationIsSafeToDeploy: + def test_it_is_non_atomic(self) -> None: + """CREATE/DROP INDEX CONCURRENTLY cannot run inside a transaction block, so + without this the migration cannot run at all. + """ + assert re.search( + r"^\s*atomic\s*=\s*False", _MIGRATION_FILE.read_text(), re.MULTILINE + ) + + def test_it_builds_and_drops_concurrently(self) -> None: + """Both directions: a plain DROP INDEX takes an ACCESS EXCLUSIVE lock, so a + rollback would block writes just as a plain build would. + """ + sql = _MIGRATION_FILE.read_text() + assert "CREATE INDEX CONCURRENTLY IF NOT EXISTS" in sql + assert "DROP INDEX CONCURRENTLY IF EXISTS" in sql + + def test_the_statement_that_runs_names_the_model_table_and_column(self) -> None: + """Everything else here reads the ``AddIndex`` state operation, which by + construction never reaches the database, or greps the source for + ``CONCURRENTLY``. The ``RunSQL`` is the only statement production executes and + its table and column were cross-checked against nothing — so the index could be + built on the wrong column while Django's model state claimed otherwise. + """ + index = _model_index() + assert index is not None + model = apps.get_model("workflow_v2", "WorkflowExecution") + expected = f"{model._meta.db_table} ({', '.join(index.fields)})" + + create = _operations()[0].database_operations[0] + assert expected in create.sql, f"expected {expected!r} in {create.sql!r}" + + def test_it_guards_against_a_leftover_invalid_index(self) -> None: + """An interrupted CONCURRENTLY build leaves an INVALID index that costs on every + write and is never read. IF NOT EXISTS would keep it while Django recorded the + migration as applied — green, and permanently slower. + """ + sql = _MIGRATION_FILE.read_text() + assert "RAISE EXCEPTION" in sql + assert "indisvalid" in sql + + def test_add_index_is_state_only(self) -> None: + """The failure mode this whole file exists for. AddIndex outside + state_operations is a real lock-taking build; inside, it only keeps Django's + model state in step so makemigrations does not re-add the index. + """ + ops = _operations() + assert len(ops) == 1 + wrapper = ops[0] + assert isinstance(wrapper, migrations.SeparateDatabaseAndState) + assert all( + isinstance(op, migrations.RunSQL) for op in wrapper.database_operations + ) + assert [type(op) for op in wrapper.state_operations] == [migrations.AddIndex] + + def test_the_migration_and_the_model_agree(self) -> None: + """Two declarations of one index; they must not drift.""" + index = _model_index() + assert index is not None + add_index = _operations()[0].state_operations[0] + assert add_index.index.name == INDEX_NAME + assert add_index.index.fields == index.fields diff --git a/workers/scheduler/dashboard_metrics_tasks.py b/workers/scheduler/dashboard_metrics_tasks.py index 90e016682a..52f7aa2c37 100644 --- a/workers/scheduler/dashboard_metrics_tasks.py +++ b/workers/scheduler/dashboard_metrics_tasks.py @@ -117,18 +117,22 @@ def _log_if_skipped(name: str, result: dict[str, Any]) -> None: @worker_task(name="dashboard_metrics.aggregate_from_sources") def dashboard_metrics_aggregate( - source_window_days: int | None = None, + tier: str | None = None, source_window_days: int | None = None ) -> dict[str, Any]: """Aggregate source tables into the hourly/daily/monthly metrics tables. - The daily reconciliation schedule dispatches a wider ``source_window_days``; - omitting it applies the backend task's own default. + Both kwargs come from the schedule row and both are optional: ``tier`` selects + which tiers to write, ``source_window_days`` widens the daily lookback for the + reconciliation pass. Omitting either applies the backend task's own default. """ - body = ( - {"source_window_days": source_window_days} - if source_window_days is not None - else None - ) + body = { + key: value + for key, value in ( + ("tier", tier), + ("source_window_days", source_window_days), + ) + if value is not None + } or None result = _call_internal(_AGGREGATE_PATH, body=body) _log_if_skipped("dashboard_metrics.aggregate_from_sources", result) return result diff --git a/workers/tests/test_dashboard_metrics_tasks.py b/workers/tests/test_dashboard_metrics_tasks.py index b78cbedb01..aae97cf63e 100644 --- a/workers/tests/test_dashboard_metrics_tasks.py +++ b/workers/tests/test_dashboard_metrics_tasks.py @@ -62,14 +62,6 @@ def test_task_is_registered_under_the_beat_name(self, name, func): assert getattr(dmt, func).name == name -# The kwargs the dashboard_metrics migrations declare on schedule rows for -# dashboard_metrics.aggregate_from_sources. Beat dispatches straight to the Django -# task, the PG scheduler dispatches to the proxy below — so a kwarg the proxy cannot -# bind raises TypeError per tick, is not covered by autoretry_for, and is dropped at -# MAX_ATTEMPTS=1. Kept in step by dashboard_metrics/tests/test_pg_periodic_task_declarations.py -# on the Django side; this is the half that lives outside Django. -_DECLARED_AGGREGATE_KWARGS = [{}, {"source_window_days": 7}] - class TestCallContract: def test_aggregate_posts_to_the_aggregate_endpoint(self): @@ -77,19 +69,30 @@ def test_aggregate_posts_to_the_aggregate_endpoint(self): dmt.dashboard_metrics_aggregate() assert call.call_args[0][0] == "v1/dashboard-metrics/aggregate/" - @pytest.mark.parametrize("kwargs", _DECLARED_AGGREGATE_KWARGS) - def test_every_scheduled_kwarg_set_binds_to_the_proxy(self, kwargs): - inspect.signature(dmt.dashboard_metrics_aggregate).bind(**kwargs) + @pytest.mark.parametrize("tier", ["hourly", "daily_monthly", "all"]) + def test_aggregate_forwards_the_tier_from_the_schedule_row(self, tier): + """UN-3974: the PG scheduler hands a row's task_kwargs over as **kwargs, so the + tier arrives here and has to reach the backend in the request body. + + This is the leg that fails quietly. Drop the forwarding and every schedule still + fires, the endpoint still returns 200, and every other test here still passes — + but both rows run the default tier, so daily and monthly quietly go back to + being recomputed every 15 minutes. + """ + with patch.object(dmt, "_call_internal", return_value={"success": True}) as call: + dmt.dashboard_metrics_aggregate(tier=tier) + assert call.call_args.kwargs["body"] == {"tier": tier} def test_aggregate_passes_the_source_window_through(self): - # The 4 AM reconciliation row carries this; dropping it here silently reverts - # the pass to the narrow 15-minute window it exists to widen. + # UN-3973: the reconciliation row carries this; dropping it here silently + # reverts the pass to the narrow window it exists to widen. with patch.object(dmt, "_call_internal", return_value={"success": True}) as call: dmt.dashboard_metrics_aggregate(source_window_days=7) assert call.call_args.kwargs["body"] == {"source_window_days": 7} - def test_aggregate_omits_body_when_no_window_given(self): - # The backend then applies the task's own default rather than one invented here. + def test_aggregate_omits_the_body_when_neither_is_given(self): + # Rows written before 0006 carry no tier kwarg; the backend default then applies, + # which is every tier rather than none. with patch.object(dmt, "_call_internal", return_value={"success": True}) as call: dmt.dashboard_metrics_aggregate() assert call.call_args.kwargs["body"] is None @@ -125,6 +128,40 @@ def test_lock_held_result_is_surfaced_not_swallowed(self, caplog): assert result["skipped"] is True +class TestTheReconciliationKwargSurvives: + """0005 declares a row against this same task path carrying source_window_days. + + The PG scheduler copies task_kwargs verbatim into the payload, so a proxy that + does not accept it raises TypeError per tick — not covered by autoretry_for, and + dropped at MAX_ATTEMPTS=1. The gap-repair pass simply never runs. + """ + + _DECLARED = [{}, {"tier": "hourly"}, {"source_window_days": 7}] + + @pytest.mark.parametrize("kwargs", _DECLARED) + def test_every_scheduled_kwarg_set_binds(self, kwargs) -> None: + inspect.signature(dmt.dashboard_metrics_aggregate).bind(**kwargs) + + def test_the_source_window_reaches_the_endpoint(self) -> None: + with patch.object(dmt, "_call_internal", return_value={}) as call: + dmt.dashboard_metrics_aggregate(source_window_days=7) + assert call.call_args.kwargs["body"] == {"source_window_days": 7} + + def test_both_kwargs_travel_together(self) -> None: + with patch.object(dmt, "_call_internal", return_value={}) as call: + dmt.dashboard_metrics_aggregate(tier="hourly", source_window_days=7) + assert call.call_args.kwargs["body"] == { + "tier": "hourly", + "source_window_days": 7, + } + + def test_omitting_both_sends_no_body(self) -> None: + # The backend then applies its own defaults rather than ones invented here. + with patch.object(dmt, "_call_internal", return_value={}) as call: + dmt.dashboard_metrics_aggregate() + assert call.call_args.kwargs["body"] is None + + class TestInternalCall: def _response(self, status_code=200, payload=None): r = MagicMock()