From fe467d837e4be72069756e9f72b7ddbf3aa0159e Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Tue, 25 Aug 2026 12:00:49 +0530 Subject: [PATCH 1/7] UN-3973 Derive monthly metrics from the daily tier, narrow source window to 2 days The dashboard aggregation widened its DAY-granularity query to the first of the previous month so monthly buckets could be summed in Python from the same rows. Every run re-read 32-62 days of source data per metric, per org, 96 times a day. Monthly is now rolled up from event_metrics_daily in one statement for all orgs, so the source queries only need the daily window. That window drops to 2 days, sized against the measured worst created_at -> terminal-status lag of ~2h. A once-daily 7-day pass reruns the same task at a wider bound to repair gaps left by cron downtime. The active-org prefilter is decoupled from the daily window and pinned at 7 days: metrics filtered on another column (hitl_completions on approved_at) can land for an org whose executions are older than the source window. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01KGZBF68CShem3pbUJM2tBc --- .../0004_add_reconciliation_task.py | 61 ++++++ backend/dashboard_metrics/tasks.py | 194 +++++++++--------- backend/dashboard_metrics/tests/test_tasks.py | 148 ++++++++++++- 3 files changed, 307 insertions(+), 96 deletions(-) create mode 100644 backend/dashboard_metrics/migrations/0004_add_reconciliation_task.py diff --git a/backend/dashboard_metrics/migrations/0004_add_reconciliation_task.py b/backend/dashboard_metrics/migrations/0004_add_reconciliation_task.py new file mode 100644 index 0000000000..0e05394687 --- /dev/null +++ b/backend/dashboard_metrics/migrations/0004_add_reconciliation_task.py @@ -0,0 +1,61 @@ +"""Data migration to schedule the daily-tier reconciliation pass. + +The 15-minute aggregation reads a narrow source window, which cannot repair +gaps left by cron downtime. This runs the same task once a day at a wider +window to backfill them. +""" + +from django.db import migrations + +RECONCILE_TASK_NAME = "dashboard_metrics_reconcile_source_window" + + +def create_reconciliation_task(apps, schema_editor): + """Create the once-daily reconciliation periodic task.""" + CrontabSchedule = apps.get_model("django_celery_beat", "CrontabSchedule") + PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask") + + # 4:00 AM UTC — clear of the 2:00 and 3:00 cleanup tasks + schedule_4am, _ = CrontabSchedule.objects.get_or_create( + minute="0", + hour="4", + day_of_week="*", + day_of_month="*", + month_of_year="*", + defaults={"timezone": "UTC"}, + ) + + PeriodicTask.objects.update_or_create( + name=RECONCILE_TASK_NAME, + defaults={ + "task": "dashboard_metrics.aggregate_from_sources", + "crontab": schedule_4am, + "queue": "dashboard_metric_events", + "kwargs": '{"source_window_days": 7}', + "enabled": True, + "description": ( + "Re-aggregate metrics over a 7 day source window to repair " + "daily-tier gaps left by cron downtime" + ), + }, + ) + + +def remove_reconciliation_task(apps, schema_editor): + """Remove the reconciliation periodic task on rollback.""" + PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask") + PeriodicTask.objects.filter(name=RECONCILE_TASK_NAME).delete() + + +class Migration(migrations.Migration): + dependencies = [ + ("dashboard_metrics", "0003_alter_eventmetricsdaily_organization_and_more"), + ("django_celery_beat", "0018_improve_crontab_helptext"), + ] + + operations = [ + migrations.RunPython( + create_reconciliation_task, + remove_reconciliation_task, + ), + ] diff --git a/backend/dashboard_metrics/tasks.py b/backend/dashboard_metrics/tasks.py index 181c985137..3c7b5fe9cd 100644 --- a/backend/dashboard_metrics/tasks.py +++ b/backend/dashboard_metrics/tasks.py @@ -8,12 +8,14 @@ import logging import time -from datetime import datetime, timedelta +from datetime import date, datetime, timedelta from typing import Any from account_v2.models import Organization from celery import shared_task from django.core.cache import cache +from django.db.models import Min, Sum +from django.db.models.functions import TruncMonth from django.db.utils import DatabaseError, OperationalError from django.utils import timezone from workflow_manager.workflow_v2.models.execution import WorkflowExecution @@ -33,6 +35,21 @@ DASHBOARD_HOURLY_METRICS_RETENTION_DAYS = 30 DASHBOARD_DAILY_METRICS_RETENTION_DAYS = 365 +# Source lookback for the daily tier. Sized against the measured worst +# created_at -> terminal-status lag of ~2h, bounded by the file processing +# time limit and the stuck-execution reaper. +DASHBOARD_SOURCE_WINDOW_DAYS = 2 + +# Wider lookback used by the once-daily reconciliation pass, which repairs +# the daily tier after cron downtime. +DASHBOARD_RECONCILE_WINDOW_DAYS = 7 + +# Lookback for the active-org prefilter. Deliberately independent of the +# source window: metrics filtered on a column other than +# WorkflowExecution.created_at (e.g. hitl_completions on approved_at) can +# land for an org whose executions are older than the source window. +DASHBOARD_ACTIVE_ORG_LOOKBACK_DAYS = 7 + def _upsert_agg(agg: dict, key: tuple, metric_type: str, value: float) -> None: """Add a value to an aggregation dict, creating the entry if needed.""" @@ -165,32 +182,47 @@ def _bulk_upsert_daily(aggregations: dict) -> int: return len(objects) -def _bulk_upsert_monthly(aggregations: dict) -> int: - """Bulk upsert monthly aggregations using INSERT ... ON CONFLICT. +def _rollup_monthly_from_daily(month_start: date) -> int: + """Derive monthly metrics by summing the daily tier from month_start onwards. - Uses _base_manager to bypass DefaultOrganizationManagerMixin. + Replaces per-org monthly queries against source tables with a single + aggregate over event_metrics_daily, which retains far more history than + the monthly window needs. + + metric_type is aggregated rather than grouped: it is not part of + unique_monthly_metric, so grouping on it could yield two rows for one + conflict target. Args: - aggregations: Dict keyed by (org_id, month_str, metric_name, project, tag) + month_start: First day of the earliest month to rebuild Returns: Number of rows upserted """ - objects = [] - for key, agg in aggregations.items(): - org_id, month_str, metric_name, project, tag = key - objects.append( - EventMetricsMonthly( - organization_id=org_id, - month=datetime.fromisoformat(month_str).date(), - metric_name=metric_name, - project=project, - tag=tag, - metric_type=agg["metric_type"], - metric_value=agg["value"], - metric_count=agg["count"], - ) + rows = ( + EventMetricsDaily._base_manager.filter(date__gte=month_start) + .annotate(month=TruncMonth("date")) + .values("organization_id", "month", "metric_name", "project", "tag") + .annotate( + value=Sum("metric_value"), + count=Sum("metric_count"), + mtype=Min("metric_type"), + ) + ) + + objects = [ + EventMetricsMonthly( + organization_id=row["organization_id"], + month=row["month"], + metric_name=row["metric_name"], + project=row["project"], + tag=row["tag"], + metric_type=row["mtype"], + metric_value=row["value"], + metric_count=row["count"], ) + for row in rows + ] if not objects: return 0 @@ -260,7 +292,9 @@ def _acquire_aggregation_lock() -> bool: retry_backoff=True, retry_backoff_max=300, ) -def aggregate_metrics_from_sources() -> dict[str, Any]: +def aggregate_metrics_from_sources( + source_window_days: int = DASHBOARD_SOURCE_WINDOW_DAYS, +) -> dict[str, Any]: """Aggregate metrics from source tables into hourly, daily, and monthly tables. This task runs periodically (every 15 minutes) to query metrics from @@ -274,8 +308,13 @@ def aggregate_metrics_from_sources() -> dict[str, Any]: Aggregation windows: - Hourly: Last 24 hours (rolling window) - - Daily: Last 7 days (ensures we capture late-arriving data) - - Monthly: Last 2 months (current + previous month) + - Daily: source_window_days (covers late-arriving data) + - Monthly: Rolled up from the daily tier, current + previous month + + Args: + source_window_days: Daily-tier source lookback. The once-daily + reconciliation pass runs the same task at + DASHBOARD_RECONCILE_WINDOW_DAYS to repair gaps after downtime. Returns: Dict with aggregation summary for all three tiers @@ -285,7 +324,7 @@ def aggregate_metrics_from_sources() -> dict[str, Any]: return {"success": True, "skipped": True, "reason": "lock_held"} try: - return _run_aggregation() + return _run_aggregation(source_window_days) finally: cache.delete(AGGREGATION_LOCK_KEY) @@ -297,18 +336,15 @@ def _aggregate_single_metric( org_id: str, hourly_start: datetime, daily_start: datetime, - monthly_start: datetime, end_date: datetime, hourly_agg: dict, daily_agg: dict, - monthly_agg: dict, extra_kwargs: dict | None = None, ) -> None: - """Run a single metric query at all 3 granularities and populate agg dicts. + """Run a single metric query at hourly and daily granularity. - Uses 2 queries instead of 3: the daily query is widened to monthly_start - and its results are split into both daily_agg and monthly_agg in Python. - This is the same pattern proven in the backfill management command. + Monthly totals are derived separately by rolling up the daily tier, so + neither query reaches further back than daily_start. """ extra_kwargs = extra_kwargs or {} @@ -324,43 +360,32 @@ def _aggregate_single_metric( key = (org_id, hour_ts.isoformat(), metric_name, "default", "") _upsert_agg(hourly_agg, key, metric_type, row["value"] or 0) - # === DAILY + MONTHLY (single query from monthly_start) === + # === DAILY === for row in query_method( org_id, - monthly_start, + daily_start, end_date, granularity=Granularity.DAY, **extra_kwargs, ): - value = row["value"] or 0 day_ts = _truncate_to_day(row["period"]) - - if day_ts >= daily_start: - key = (org_id, day_ts.date().isoformat(), metric_name, "default", "") - _upsert_agg(daily_agg, key, metric_type, value) - - month_key = _truncate_to_month(row["period"]).date().isoformat() - key = (org_id, month_key, metric_name, "default", "") - _upsert_agg(monthly_agg, key, metric_type, value) + key = (org_id, day_ts.date().isoformat(), metric_name, "default", "") + _upsert_agg(daily_agg, key, metric_type, row["value"] or 0) def _aggregate_llm_combined( org_id: str, hourly_start: datetime, daily_start: datetime, - monthly_start: datetime, end_date: datetime, hourly_agg: dict, daily_agg: dict, - monthly_agg: dict, llm_combined_fields: dict, ) -> None: - """Run the combined LLM metrics query at all granularities. + """Run the combined LLM metrics query at hourly and daily granularity. - Issues 2 queries total (hourly + daily/monthly) instead of 3. - The DAY-granularity query is widened to monthly_start and results are - split into daily_agg (recent rows) and monthly_agg (all rows bucketed - by month) in Python. Same pattern as _aggregate_single_metric. + Issues 2 queries covering 4 metrics. Same windowing as + _aggregate_single_metric — monthly is derived from the daily tier. """ # === HOURLY (last 24h) === for row in MetricsQueryService.get_llm_metrics_combined( @@ -374,28 +399,22 @@ def _aggregate_llm_combined( key = (org_id, ts_str, metric_name, "default", "") _upsert_agg(hourly_agg, key, metric_type, row[field] or 0) - # === DAILY + MONTHLY (single query from monthly_start) === + # === DAILY === for row in MetricsQueryService.get_llm_metrics_combined( org_id, - monthly_start, + daily_start, end_date, granularity=Granularity.DAY, ): - day_ts = _truncate_to_day(row["period"]) - month_key = _truncate_to_month(row["period"]).date().isoformat() - + day_str = _truncate_to_day(row["period"]).date().isoformat() for field, (metric_name, metric_type) in llm_combined_fields.items(): - value = row[field] or 0 + key = (org_id, day_str, metric_name, "default", "") + _upsert_agg(daily_agg, key, metric_type, row[field] or 0) - if day_ts >= daily_start: - key = (org_id, day_ts.date().isoformat(), metric_name, "default", "") - _upsert_agg(daily_agg, key, metric_type, value) - key = (org_id, month_key, metric_name, "default", "") - _upsert_agg(monthly_agg, key, metric_type, value) - - -def _run_aggregation() -> dict[str, Any]: +def _run_aggregation( + source_window_days: int = DASHBOARD_SOURCE_WINDOW_DAYS, +) -> dict[str, Any]: """Execute the actual aggregation logic. Separated from the task function to keep the lock management clean. @@ -404,25 +423,13 @@ def _run_aggregation() -> dict[str, Any]: # Query windows for each granularity # - Hourly: Last 24 hours (rolling window, matches retention of 30 days) - # - Daily: Last 7 days (ensures we capture late-arriving data) - # - Monthly: Last 2 months (current + previous, ensures month transitions are captured) + # - Daily: source_window_days of source data + # - Monthly: rolled up from the daily tier, current + previous month hourly_start = end_date - timedelta(hours=24) - daily_start = _truncate_to_day(end_date - timedelta(days=7)) - # Include previous month to handle month boundaries - if end_date.month == 1: - monthly_start = end_date.replace( - year=end_date.year - 1, - month=12, - day=1, - hour=0, - minute=0, - second=0, - microsecond=0, - ) - else: - monthly_start = end_date.replace( - month=end_date.month - 1, day=1, hour=0, minute=0, second=0, microsecond=0 - ) + daily_start = _truncate_to_day(end_date - timedelta(days=source_window_days)) + monthly_start = _truncate_to_month( + _truncate_to_month(end_date) - timedelta(days=1) + ).date() # Metric definitions: (name, query_method, is_histogram) # Note: llm_calls, challenges, summarization_calls, and llm_usage are @@ -459,15 +466,13 @@ def _run_aggregation() -> dict[str, Any]: "orgs_processed": 0, } - # Pre-filter to orgs with recent activity to reduce DB load. - # Uses daily_start (7 days) instead of monthly_start (2 months) because: - # - Hourly/daily queries only need recent data (24h / 7d windows) - # - Monthly totals for dormant orgs were already written by previous - # runs when the org was active — re-running just overwrites same values - # - This avoids 28 queries per dormant org that had activity 2-8 weeks ago + # Pre-filter to orgs with recent activity to reduce DB load. Kept wider + # than the source window because some metrics are filtered on a column + # other than WorkflowExecution.created_at (hitl_completions uses + # approved_at) and can land for an org whose executions are older. active_org_ids = set( WorkflowExecution.objects.filter( - created_at__gte=daily_start, + created_at__gte=end_date - timedelta(days=DASHBOARD_ACTIVE_ORG_LOOKBACK_DAYS), ) .values_list("workflow__organization_id", flat=True) .distinct() @@ -499,7 +504,6 @@ def _run_aggregation() -> dict[str, Any]: org_identifier = org.organization_id # Pre-resolved for PageUsage queries hourly_agg: dict[tuple, dict] = {} daily_agg: dict[tuple, dict] = {} - monthly_agg: dict[tuple, dict] = {} try: for metric_name, query_method, is_histogram in metric_configs: @@ -519,11 +523,9 @@ def _run_aggregation() -> dict[str, Any]: org_id, hourly_start, daily_start, - monthly_start, end_date, hourly_agg, daily_agg, - monthly_agg, extra_kwargs, ) except Exception: @@ -536,33 +538,35 @@ def _run_aggregation() -> dict[str, Any]: org_id, hourly_start, daily_start, - monthly_start, end_date, hourly_agg, daily_agg, - monthly_agg, llm_combined_fields, ) except Exception: logger.exception("Error querying combined LLM metrics for org %s", org_id) stats["errors"] += 1 - # Bulk upsert all three tiers (single INSERT...ON CONFLICT each) + # Bulk upsert both tiers (single INSERT...ON CONFLICT each) if hourly_agg: stats["hourly"]["upserted"] += _bulk_upsert_hourly(hourly_agg) if daily_agg: stats["daily"]["upserted"] += _bulk_upsert_daily(daily_agg) - if monthly_agg: - stats["monthly"]["upserted"] += _bulk_upsert_monthly(monthly_agg) - stats["orgs_processed"] += 1 except Exception: logger.exception("Error processing org %s", org_id) stats["errors"] += 1 + # Monthly is derived from the daily tier in one statement for all orgs + try: + stats["monthly"]["upserted"] = _rollup_monthly_from_daily(monthly_start) + except Exception: + logger.exception("Error rolling up monthly metrics from %s", monthly_start) + stats["errors"] += 1 + logger.info( f"Aggregation completed: {stats['orgs_processed']} orgs, " f"hourly={stats['hourly']['upserted']}, " diff --git a/backend/dashboard_metrics/tests/test_tasks.py b/backend/dashboard_metrics/tests/test_tasks.py index 03ef136508..c80c38303c 100644 --- a/backend/dashboard_metrics/tests/test_tasks.py +++ b/backend/dashboard_metrics/tests/test_tasks.py @@ -1,6 +1,7 @@ """Unit tests for Dashboard Metrics Celery tasks.""" -from datetime import datetime, timedelta +from datetime import date, datetime, timedelta +from unittest.mock import patch from django.test import TestCase from django.utils import timezone @@ -9,12 +10,18 @@ from dashboard_metrics.models import ( EventMetricsDaily, EventMetricsHourly, + EventMetricsMonthly, MetricType, ) from dashboard_metrics.tasks import ( + DASHBOARD_RECONCILE_WINDOW_DAYS, + DASHBOARD_SOURCE_WINDOW_DAYS, + _rollup_monthly_from_daily, + _run_aggregation, _truncate_to_day, _truncate_to_hour, _truncate_to_month, + aggregate_metrics_from_sources, cleanup_daily_metrics, cleanup_hourly_metrics, ) @@ -198,3 +205,142 @@ def test_cleanup_no_records_to_delete(self): assert result["success"] is True assert result["deleted"] == 0 + + +class TestMonthlyRollup(TestCase): + """Tests for deriving monthly metrics from the daily tier.""" + + def setUp(self): + """Set up test fixtures.""" + self.org = Organization.objects.create( + organization_id="rollup-org", name="rollup-org", display_name="Rollup Org" + ) + + def _daily(self, day, value, count=1, metric_type=MetricType.COUNTER): + """Create a daily metric row for the fixture org.""" + EventMetricsDaily.objects.create( + organization=self.org, + date=day, + metric_name="documents_processed", + metric_type=metric_type, + metric_value=value, + metric_count=count, + project="default", + ) + + def _monthly_rows(self): + """Read back monthly rows ordered by month.""" + return list(EventMetricsMonthly._base_manager.order_by("month")) + + def test_sums_daily_rows_into_month_bucket(self): + """Daily rows within a month sum into a single monthly row.""" + self._daily(date(2024, 3, 5), value=10, count=2) + self._daily(date(2024, 3, 18), value=32, count=4) + + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == 1 + + rows = self._monthly_rows() + assert len(rows) == 1 + assert rows[0].month == date(2024, 3, 1) + assert rows[0].metric_value == 42 + assert rows[0].metric_count == 6 + + def test_month_boundary_keeps_months_separate(self): + """Rows spanning the 1st land in two months without bleeding.""" + self._daily(date(2024, 1, 30), value=5) + self._daily(date(2024, 1, 31), value=7) + self._daily(date(2024, 2, 1), value=100) + self._daily(date(2024, 2, 2), value=200) + + assert _rollup_monthly_from_daily(date(2024, 1, 1)) == 2 + + rows = self._monthly_rows() + assert [r.month for r in rows] == [date(2024, 1, 1), date(2024, 2, 1)] + assert [r.metric_value for r in rows] == [12, 300] + + def test_excludes_months_before_the_window(self): + """Daily rows older than month_start are not rolled up.""" + self._daily(date(2023, 12, 15), value=999) + self._daily(date(2024, 1, 15), value=5) + + assert _rollup_monthly_from_daily(date(2024, 1, 1)) == 1 + + rows = self._monthly_rows() + assert len(rows) == 1 + assert rows[0].month == date(2024, 1, 1) + + def test_rerun_overwrites_instead_of_accumulating(self): + """A second rollup replaces the monthly total rather than doubling it.""" + self._daily(date(2024, 3, 5), value=10, count=2) + _rollup_monthly_from_daily(date(2024, 3, 1)) + + self._daily(date(2024, 3, 6), value=5, count=1) + _rollup_monthly_from_daily(date(2024, 3, 1)) + + rows = self._monthly_rows() + assert len(rows) == 1 + assert rows[0].metric_value == 15 + assert rows[0].metric_count == 3 + + def test_mixed_metric_type_within_a_month_yields_one_row(self): + """metric_type is aggregated, so it cannot split one conflict target.""" + self._daily(date(2024, 3, 5), value=10, metric_type=MetricType.HISTOGRAM) + self._daily(date(2024, 3, 6), value=5, metric_type=MetricType.COUNTER) + + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == 1 + + rows = self._monthly_rows() + assert len(rows) == 1 + assert rows[0].metric_value == 15 + + def test_no_daily_rows_upserts_nothing(self): + """An empty daily tier is a no-op, not an error.""" + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == 0 + assert not self._monthly_rows() + + +class TestSourceWindow(TestCase): + """Tests for the per-run source window and the reconciliation pass.""" + + def setUp(self): + """Set up test fixtures.""" + self.org = Organization.objects.create( + organization_id="window-org", name="window-org", display_name="Window Org" + ) + + def _run_with_active_org(self, **kwargs): + """Run aggregation with the active-org prefilter stubbed to the fixture org.""" + with patch("dashboard_metrics.tasks.WorkflowExecution") as mock_execution: + prefilter = mock_execution.objects.filter.return_value + prefilter.values_list.return_value.distinct.return_value = [self.org.id] + return _run_aggregation(**kwargs) + + def test_default_window_bounds_the_daily_query(self): + """The per-run daily window is DASHBOARD_SOURCE_WINDOW_DAYS wide.""" + result = self._run_with_active_org() + + expected = _truncate_to_day( + timezone.now() - timedelta(days=DASHBOARD_SOURCE_WINDOW_DAYS) + ) + assert result["period"]["daily"]["start"] == expected.isoformat() + + def test_reconciliation_window_widens_the_daily_query(self): + """The reconciliation pass reaches further back on the same code path.""" + result = self._run_with_active_org( + source_window_days=DASHBOARD_RECONCILE_WINDOW_DAYS + ) + + expected = _truncate_to_day( + timezone.now() - timedelta(days=DASHBOARD_RECONCILE_WINDOW_DAYS) + ) + assert result["period"]["daily"]["start"] == expected.isoformat() + + def test_task_passes_the_window_through(self): + """The scheduled task forwards its kwarg, defaulting to the per-run window.""" + with patch("dashboard_metrics.tasks._run_aggregation") as mock_run: + aggregate_metrics_from_sources() + mock_run.assert_called_once_with(DASHBOARD_SOURCE_WINDOW_DAYS) + + with patch("dashboard_metrics.tasks._run_aggregation") as mock_run: + aggregate_metrics_from_sources(source_window_days=7) + mock_run.assert_called_once_with(7) From 3ea08b721360ec5fc90f347bfbe91e11912f51f6 Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Tue, 25 Aug 2026 20:26:30 +0530 Subject: [PATCH 2/7] UN-3973 Address Sonar and Greptile review findings Sonar: - S117: rename apps.get_model() locals in 0004 to snake_case - S3776: cut _run_aggregation cognitive complexity from 22 by hoisting the static metric config tables to module level and extracting the per-org body, the active-org prefilter and the result shape into helpers Greptile: - Monthly rows in the rebuilt window whose daily rows are gone are now deleted alongside the upsert, so the two tiers cannot disagree. An empty daily tier still short-circuits, so a wiped tier cannot cascade into deleting monthly history. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01KGZBF68CShem3pbUJM2tBc --- .../0004_add_reconciliation_task.py | 12 +- backend/dashboard_metrics/tasks.py | 342 +++++++++++------- backend/dashboard_metrics/tests/test_tasks.py | 26 ++ 3 files changed, 249 insertions(+), 131 deletions(-) diff --git a/backend/dashboard_metrics/migrations/0004_add_reconciliation_task.py b/backend/dashboard_metrics/migrations/0004_add_reconciliation_task.py index 0e05394687..09667e867f 100644 --- a/backend/dashboard_metrics/migrations/0004_add_reconciliation_task.py +++ b/backend/dashboard_metrics/migrations/0004_add_reconciliation_task.py @@ -12,11 +12,11 @@ def create_reconciliation_task(apps, schema_editor): """Create the once-daily reconciliation periodic task.""" - CrontabSchedule = apps.get_model("django_celery_beat", "CrontabSchedule") - PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask") + crontab_model = apps.get_model("django_celery_beat", "CrontabSchedule") + periodic_task_model = apps.get_model("django_celery_beat", "PeriodicTask") # 4:00 AM UTC — clear of the 2:00 and 3:00 cleanup tasks - schedule_4am, _ = CrontabSchedule.objects.get_or_create( + schedule_4am, _ = crontab_model.objects.get_or_create( minute="0", hour="4", day_of_week="*", @@ -25,7 +25,7 @@ def create_reconciliation_task(apps, schema_editor): defaults={"timezone": "UTC"}, ) - PeriodicTask.objects.update_or_create( + periodic_task_model.objects.update_or_create( name=RECONCILE_TASK_NAME, defaults={ "task": "dashboard_metrics.aggregate_from_sources", @@ -43,8 +43,8 @@ def create_reconciliation_task(apps, schema_editor): def remove_reconciliation_task(apps, schema_editor): """Remove the reconciliation periodic task on rollback.""" - PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask") - PeriodicTask.objects.filter(name=RECONCILE_TASK_NAME).delete() + periodic_task_model = apps.get_model("django_celery_beat", "PeriodicTask") + periodic_task_model.objects.filter(name=RECONCILE_TASK_NAME).delete() class Migration(migrations.Migration): diff --git a/backend/dashboard_metrics/tasks.py b/backend/dashboard_metrics/tasks.py index 3c7b5fe9cd..d01b9b5418 100644 --- a/backend/dashboard_metrics/tasks.py +++ b/backend/dashboard_metrics/tasks.py @@ -14,6 +14,7 @@ from account_v2.models import Organization from celery import shared_task from django.core.cache import cache +from django.db import transaction from django.db.models import Min, Sum from django.db.models.functions import TruncMonth from django.db.utils import DatabaseError, OperationalError @@ -182,12 +183,48 @@ def _bulk_upsert_daily(aggregations: dict) -> int: return len(objects) +def _delete_orphan_monthly(month_start: date, fresh_keys: set[tuple]) -> int: + """Drop monthly rows in the window that the rollup no longer produces. + + Monthly is a pure derivation of the daily tier, so a key whose daily rows + have gone (a deleted workflow cascades to its executions) must not survive + as a stale total. + + Args: + month_start: First day of the earliest month being rebuilt + fresh_keys: Keys the current rollup produced + + Returns: + Number of rows deleted + """ + stale_pks = [ + row["pk"] + for row in EventMetricsMonthly._base_manager.filter( + month__gte=month_start + ).values("pk", "organization_id", "month", "metric_name", "project", "tag") + if ( + row["organization_id"], + row["month"], + row["metric_name"], + row["project"], + row["tag"], + ) + not in fresh_keys + ] + if not stale_pks: + return 0 + + deleted, _ = EventMetricsMonthly._base_manager.filter(pk__in=stale_pks).delete() + return deleted + + def _rollup_monthly_from_daily(month_start: date) -> int: """Derive monthly metrics by summing the daily tier from month_start onwards. Replaces per-org monthly queries against source tables with a single aggregate over event_metrics_daily, which retains far more history than - the monthly window needs. + the monthly window needs. Rows in the window that the daily tier no longer + backs are removed, so the two tiers cannot disagree. metric_type is aggregated rather than grouped: it is not part of unique_monthly_metric, so grouping on it could yield two rows for one @@ -224,15 +261,24 @@ def _rollup_monthly_from_daily(month_start: date) -> int: for row in rows ] + # An empty daily tier means the source of truth is gone, not that every + # month is genuinely zero, so leave the existing rows alone. if not objects: return 0 - EventMetricsMonthly._base_manager.bulk_create( - objects, - update_conflicts=True, - unique_fields=["organization", "month", "metric_name", "project", "tag"], - update_fields=["metric_type", "metric_value", "metric_count"], - ) + fresh_keys = { + (o.organization_id, o.month, o.metric_name, o.project, o.tag) for o in objects + } + + with transaction.atomic(): + EventMetricsMonthly._base_manager.bulk_create( + objects, + update_conflicts=True, + unique_fields=["organization", "month", "metric_name", "project", "tag"], + update_fields=["metric_type", "metric_value", "metric_count"], + ) + _delete_orphan_monthly(month_start, fresh_keys) + return len(objects) @@ -412,6 +458,156 @@ def _aggregate_llm_combined( _upsert_agg(daily_agg, key, metric_type, row[field] or 0) +# Metric definitions: (name, query_method, is_histogram) +# Note: llm_calls, challenges, summarization_calls, and llm_usage are +# handled separately via get_llm_metrics_combined (1 query instead of 4). +METRIC_CONFIGS = [ + ("documents_processed", MetricsQueryService.get_documents_processed, False), + ("pages_processed", MetricsQueryService.get_pages_processed, True), + ("deployed_api_requests", MetricsQueryService.get_deployed_api_requests, False), + ("etl_pipeline_executions", MetricsQueryService.get_etl_pipeline_executions, False), + ("prompt_executions", MetricsQueryService.get_prompt_executions, False), + ("failed_pages", MetricsQueryService.get_failed_pages, True), + ("hitl_reviews", MetricsQueryService.get_hitl_reviews, False), + ("hitl_completions", MetricsQueryService.get_hitl_completions, False), +] + +# LLM metrics combined via conditional aggregation (4 metrics in 1 query). +# Maps combined query field -> (metric_name, metric_type) +LLM_COMBINED_FIELDS = { + "llm_calls": ("llm_calls", MetricType.COUNTER), + "challenges": ("challenges", MetricType.COUNTER), + "summarization_calls": ("summarization_calls", MetricType.COUNTER), + "llm_usage": ("llm_usage", MetricType.HISTOGRAM), +} + + +def _collect_org_metrics( + org: Organization, + hourly_start: datetime, + daily_start: datetime, + end_date: datetime, +) -> tuple[dict, dict, int]: + """Query every metric for one organization into hourly/daily aggregates. + + A failing metric is logged and counted, leaving the rest to proceed. + + Returns: + Tuple of (hourly aggregations, daily aggregations, error count) + """ + org_id = str(org.id) + hourly_agg: dict[tuple, dict] = {} + daily_agg: dict[tuple, dict] = {} + errors = 0 + + for metric_name, query_method, is_histogram in METRIC_CONFIGS: + metric_type = MetricType.HISTOGRAM if is_histogram else MetricType.COUNTER + # Pre-resolved identifier spares PageUsage an Organization lookup per call. + extra_kwargs = ( + {"org_identifier": org.organization_id} + if metric_name == "pages_processed" + else {} + ) + try: + _aggregate_single_metric( + query_method, + metric_name, + metric_type, + org_id, + hourly_start, + daily_start, + end_date, + hourly_agg, + daily_agg, + extra_kwargs, + ) + except Exception: + logger.exception("Error querying %s for org %s", metric_name, org_id) + errors += 1 + + try: + _aggregate_llm_combined( + org_id, + hourly_start, + daily_start, + end_date, + hourly_agg, + daily_agg, + LLM_COMBINED_FIELDS, + ) + except Exception: + logger.exception("Error querying combined LLM metrics for org %s", org_id) + errors += 1 + + return hourly_agg, daily_agg, errors + + +def _aggregate_org( + org: Organization, + hourly_start: datetime, + daily_start: datetime, + end_date: datetime, + stats: dict[str, Any], +) -> None: + """Aggregate one organization and upsert its hourly and daily tiers.""" + hourly_agg, daily_agg, errors = _collect_org_metrics( + org, hourly_start, daily_start, end_date + ) + stats["errors"] += errors + + # Bulk upsert both tiers (single INSERT...ON CONFLICT each) + if hourly_agg: + stats["hourly"]["upserted"] += _bulk_upsert_hourly(hourly_agg) + + if daily_agg: + stats["daily"]["upserted"] += _bulk_upsert_daily(daily_agg) + + stats["orgs_processed"] += 1 + + +def _active_org_ids(end_date: datetime) -> set: + """Organizations with recent execution activity. + + Deliberately wider than the source window: metrics filtered on a column + other than WorkflowExecution.created_at (hitl_completions uses approved_at) + can land for an org whose executions are older. + """ + return set( + WorkflowExecution.objects.filter( + created_at__gte=end_date - timedelta(days=DASHBOARD_ACTIVE_ORG_LOOKBACK_DAYS), + ) + .values_list("workflow__organization_id", flat=True) + .distinct() + ) + + +def _build_result( + stats: dict[str, Any], + hourly_start: datetime, + daily_start: datetime, + monthly_start: date, + end_date: datetime, + skipped_reason: str | None = None, +) -> dict[str, Any]: + """Shape the task's return value from the accumulated stats.""" + result = { + "success": True, + "organizations_processed": stats["orgs_processed"], + "hourly": stats["hourly"], + "daily": stats["daily"], + "monthly": stats["monthly"], + "errors": stats["errors"], + "period": { + "hourly": {"start": hourly_start.isoformat(), "end": end_date.isoformat()}, + "daily": {"start": daily_start.isoformat(), "end": end_date.isoformat()}, + "monthly": {"start": monthly_start.isoformat(), "end": end_date.isoformat()}, + }, + } + if skipped_reason: + result["skipped_reason"] = skipped_reason + return result + + def _run_aggregation( source_window_days: int = DASHBOARD_SOURCE_WINDOW_DAYS, ) -> dict[str, Any]: @@ -431,33 +627,6 @@ def _run_aggregation( _truncate_to_month(end_date) - timedelta(days=1) ).date() - # Metric definitions: (name, query_method, is_histogram) - # Note: llm_calls, challenges, summarization_calls, and llm_usage are - # handled separately via get_llm_metrics_combined (1 query instead of 4). - metric_configs = [ - ("documents_processed", MetricsQueryService.get_documents_processed, False), - ("pages_processed", MetricsQueryService.get_pages_processed, True), - ("deployed_api_requests", MetricsQueryService.get_deployed_api_requests, False), - ( - "etl_pipeline_executions", - MetricsQueryService.get_etl_pipeline_executions, - False, - ), - ("prompt_executions", MetricsQueryService.get_prompt_executions, False), - ("failed_pages", MetricsQueryService.get_failed_pages, True), - ("hitl_reviews", MetricsQueryService.get_hitl_reviews, False), - ("hitl_completions", MetricsQueryService.get_hitl_completions, False), - ] - - # LLM metrics combined via conditional aggregation (4 metrics in 1 query). - # Maps combined query field -> (metric_name, metric_type) - llm_combined_fields = { - "llm_calls": ("llm_calls", MetricType.COUNTER), - "challenges": ("challenges", MetricType.COUNTER), - "summarization_calls": ("summarization_calls", MetricType.COUNTER), - "llm_usage": ("llm_usage", MetricType.HISTOGRAM), - } - stats = { "hourly": {"upserted": 0}, "daily": {"upserted": 0}, @@ -466,98 +635,33 @@ def _run_aggregation( "orgs_processed": 0, } - # Pre-filter to orgs with recent activity to reduce DB load. Kept wider - # than the source window because some metrics are filtered on a column - # other than WorkflowExecution.created_at (hitl_completions uses - # approved_at) and can land for an org whose executions are older. - active_org_ids = set( - WorkflowExecution.objects.filter( - created_at__gte=end_date - timedelta(days=DASHBOARD_ACTIVE_ORG_LOOKBACK_DAYS), - ) - .values_list("workflow__organization_id", flat=True) - .distinct() - ) - total_orgs = Organization.objects.count() + # Pre-filter to orgs with recent activity to reduce DB load. + active_org_ids = _active_org_ids(end_date) logger.info( "Aggregation: %d active orgs out of %d total", len(active_org_ids), - total_orgs, + Organization.objects.count(), ) if not active_org_ids: - return { - "success": True, - "organizations_processed": 0, - "hourly": stats["hourly"], - "daily": stats["daily"], - "monthly": stats["monthly"], - "errors": 0, - "skipped_reason": "no_active_orgs", - } + return _build_result( + stats, + hourly_start, + daily_start, + monthly_start, + end_date, + skipped_reason="no_active_orgs", + ) organizations = Organization.objects.filter(id__in=active_org_ids).only( "id", "organization_id" ) for org in organizations: - org_id = str(org.id) - org_identifier = org.organization_id # Pre-resolved for PageUsage queries - hourly_agg: dict[tuple, dict] = {} - daily_agg: dict[tuple, dict] = {} - try: - for metric_name, query_method, is_histogram in metric_configs: - metric_type = MetricType.HISTOGRAM if is_histogram else MetricType.COUNTER - - # Pass org_identifier to PageUsage-based metrics to - # avoid redundant Organization lookups per call. - extra_kwargs = {} - if metric_name == "pages_processed": - extra_kwargs["org_identifier"] = org_identifier - - try: - _aggregate_single_metric( - query_method, - metric_name, - metric_type, - org_id, - hourly_start, - daily_start, - end_date, - hourly_agg, - daily_agg, - extra_kwargs, - ) - except Exception: - logger.exception("Error querying %s for org %s", metric_name, org_id) - stats["errors"] += 1 - - # Combined LLM metrics: 1 query per granularity instead of 4 - try: - _aggregate_llm_combined( - org_id, - hourly_start, - daily_start, - end_date, - hourly_agg, - daily_agg, - llm_combined_fields, - ) - except Exception: - logger.exception("Error querying combined LLM metrics for org %s", org_id) - stats["errors"] += 1 - - # Bulk upsert both tiers (single INSERT...ON CONFLICT each) - if hourly_agg: - stats["hourly"]["upserted"] += _bulk_upsert_hourly(hourly_agg) - - if daily_agg: - stats["daily"]["upserted"] += _bulk_upsert_daily(daily_agg) - - stats["orgs_processed"] += 1 - + _aggregate_org(org, hourly_start, daily_start, end_date, stats) except Exception: - logger.exception("Error processing org %s", org_id) + logger.exception("Error processing org %s", org.id) stats["errors"] += 1 # Monthly is derived from the daily tier in one statement for all orgs @@ -575,19 +679,7 @@ def _run_aggregation( f"errors={stats['errors']}" ) - return { - "success": True, - "organizations_processed": stats["orgs_processed"], - "hourly": stats["hourly"], - "daily": stats["daily"], - "monthly": stats["monthly"], - "errors": stats["errors"], - "period": { - "hourly": {"start": hourly_start.isoformat(), "end": end_date.isoformat()}, - "daily": {"start": daily_start.isoformat(), "end": end_date.isoformat()}, - "monthly": {"start": monthly_start.isoformat(), "end": end_date.isoformat()}, - }, - } + return _build_result(stats, hourly_start, daily_start, monthly_start, end_date) @shared_task( diff --git a/backend/dashboard_metrics/tests/test_tasks.py b/backend/dashboard_metrics/tests/test_tasks.py index c80c38303c..682a0f7246 100644 --- a/backend/dashboard_metrics/tests/test_tasks.py +++ b/backend/dashboard_metrics/tests/test_tasks.py @@ -298,6 +298,32 @@ def test_no_daily_rows_upserts_nothing(self): assert _rollup_monthly_from_daily(date(2024, 3, 1)) == 0 assert not self._monthly_rows() + def test_monthly_row_is_dropped_once_its_daily_rows_are_gone(self): + """A month whose daily rows were deleted must not keep a stale total.""" + self._daily(date(2024, 3, 5), value=10) + self._daily(date(2024, 4, 5), value=7) + _rollup_monthly_from_daily(date(2024, 3, 1)) + assert len(self._monthly_rows()) == 2 + + EventMetricsDaily._base_manager.filter(date=date(2024, 3, 5)).delete() + _rollup_monthly_from_daily(date(2024, 3, 1)) + + rows = self._monthly_rows() + assert len(rows) == 1 + assert rows[0].month == date(2024, 4, 1) + + def test_months_before_the_window_are_left_alone(self): + """Orphan cleanup must not reach outside the rebuilt window.""" + self._daily(date(2024, 1, 10), value=99) + _rollup_monthly_from_daily(date(2024, 1, 1)) + EventMetricsDaily._base_manager.all().delete() + + self._daily(date(2024, 3, 5), value=10) + _rollup_monthly_from_daily(date(2024, 3, 1)) + + months = [row.month for row in self._monthly_rows()] + assert months == [date(2024, 1, 1), date(2024, 3, 1)] + class TestSourceWindow(TestCase): """Tests for the per-run source window and the reconciliation pass.""" From 395393465b95bd8ca97c4a17e702ac548769fd7d Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Fri, 28 Aug 2026 19:30:01 +0530 Subject: [PATCH 3/7] UN-3973 Add tests covering the source window, rollup SQL and reconciliation schedule Closes the acceptance criteria that had no automated check: - the monthly rollup issues no source-table SQL, asserted by capturing the queries it actually sends - the window ladder at 2 / 7 / 62 days, including a row that finishes after the narrow window has moved past its created_at and so never re-enters it - the reconciliation schedule row, its idempotency and its reverse The schedule tests call the migration's function directly. The suite runs with --no-migrations, so data migrations never execute and asserting on the beat row would fail regardless of the migration being correct. Also moves the dotenv load in settings/base.py above the Celery block. CELERY_BROKER_BASE_URL, _USER and _PASS were read above it, so they could not be supplied by an env file at all and had to be ambient. Ambient values still take precedence, so deployed behaviour is unchanged. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01KGZBF68CShem3pbUJM2tBc --- backend/backend/settings/base.py | 8 +- backend/dashboard_metrics/tests/test_tasks.py | 160 ++++++++++++++++++ 2 files changed, 164 insertions(+), 4 deletions(-) diff --git a/backend/backend/settings/base.py b/backend/backend/settings/base.py index d14f87b304..f0dbafe6d2 100644 --- a/backend/backend/settings/base.py +++ b/backend/backend/settings/base.py @@ -55,6 +55,10 @@ def get_required_setting(setting_key: str, default: str | None = None) -> str | # Load default log from env DEFAULT_LOG_LEVEL = os.environ.get("DEFAULT_LOG_LEVEL", "INFO") +ENV_FILE = find_dotenv() +if ENV_FILE: + load_dotenv(ENV_FILE) + # Celery Broker Configuration CELERY_BROKER_BASE_URL = get_required_setting("CELERY_BROKER_BASE_URL") CELERY_BROKER_USER = get_required_setting("CELERY_BROKER_USER") @@ -65,10 +69,6 @@ def get_required_setting(setting_key: str, default: str | None = None) -> str | ) ) -ENV_FILE = find_dotenv() -if ENV_FILE: - load_dotenv(ENV_FILE) - # Loading environment variables WORKFLOW_ACTION_EXPIRATION_TIME_IN_SECOND = os.environ.get( diff --git a/backend/dashboard_metrics/tests/test_tasks.py b/backend/dashboard_metrics/tests/test_tasks.py index 682a0f7246..6cb9316d1c 100644 --- a/backend/dashboard_metrics/tests/test_tasks.py +++ b/backend/dashboard_metrics/tests/test_tasks.py @@ -1,10 +1,16 @@ """Unit tests for Dashboard Metrics Celery tasks.""" +import json from datetime import date, datetime, timedelta +from importlib import import_module from unittest.mock import patch +from django.apps import apps +from django.db import connection from django.test import TestCase +from django.test.utils import CaptureQueriesContext from django.utils import timezone +from django_celery_beat.models import PeriodicTask from account_v2.models import Organization from dashboard_metrics.models import ( @@ -13,6 +19,10 @@ EventMetricsMonthly, MetricType, ) +from workflow_manager.file_execution.models import WorkflowFileExecution +from workflow_manager.workflow_v2.enums import ExecutionStatus +from workflow_manager.workflow_v2.models.execution import WorkflowExecution +from workflow_manager.workflow_v2.models.workflow import Workflow from dashboard_metrics.tasks import ( DASHBOARD_RECONCILE_WINDOW_DAYS, DASHBOARD_SOURCE_WINDOW_DAYS, @@ -325,6 +335,37 @@ def test_months_before_the_window_are_left_alone(self): assert months == [date(2024, 1, 1), date(2024, 3, 1)] +class TestRollupQueryShape(TestCase): + """The monthly rollup must not read the raw source tables.""" + + def test_monthly_rollup_never_touches_source_tables(self): + """This is the saving: monthly reads the daily tier and nothing else.""" + EventMetricsDaily._base_manager.create( + organization=Organization.objects.create( + organization_id="shape-org", name="shape", display_name="Shape" + ), + date=date(2024, 3, 5), + metric_name="documents_processed", + metric_type=MetricType.COUNTER, + metric_value=10, + metric_count=2, + project="default", + tag="", + ) + + with CaptureQueriesContext(connection) as captured: + _rollup_monthly_from_daily(date(2024, 3, 1)) + + sql = " ".join(q["sql"] for q in captured.captured_queries).lower() + assert "event_metrics_daily" in sql + for source_table in ( + "workflow_file_execution", + "workflow_execution", + "page_usage", + ): + assert source_table not in sql, f"monthly rollup read {source_table}" + + class TestSourceWindow(TestCase): """Tests for the per-run source window and the reconciliation pass.""" @@ -370,3 +411,122 @@ def test_task_passes_the_window_through(self): with patch("dashboard_metrics.tasks._run_aggregation") as mock_run: aggregate_metrics_from_sources(source_window_days=7) mock_run.assert_called_once_with(7) + + def _seed_file( + self, days_ago: int, status: ExecutionStatus = ExecutionStatus.COMPLETED + ) -> date: + """Seed one file execution dated days_ago, return its date.""" + workflow = Workflow.objects.create( + workflow_name=f"recon-wf-{days_ago}", organization=self.org + ) + execution = WorkflowExecution.objects.create( + workflow=workflow, status=ExecutionStatus.COMPLETED + ) + file_execution = WorkflowFileExecution.objects.create( + workflow_execution=execution, + file_name="a.pdf", + status=status.value, + ) + + stamp = timezone.now() - timedelta(days=days_ago) + # created_at is auto_now_add; a queryset update is what bypasses it + WorkflowFileExecution.objects.filter(pk=file_execution.pk).update( + created_at=stamp + ) + WorkflowExecution.objects.filter(pk=execution.pk).update(created_at=stamp) + return stamp.date() + + def test_reconciliation_recovers_a_day_the_narrow_window_missed(self): + """A row outside the per-run window is picked up by the wider pass.""" + day = self._seed_file(days_ago=5) + + _run_aggregation() + assert not EventMetricsDaily._base_manager.filter(date=day).exists() + + result = _run_aggregation(source_window_days=DASHBOARD_RECONCILE_WINDOW_DAYS) + + row = EventMetricsDaily._base_manager.get( + date=day, metric_name="documents_processed" + ) + assert row.metric_value == 1 + assert result["errors"] == 0 + + def test_late_terminal_status_does_not_re_enter_the_narrow_window(self): + """Finishing after the window moved on does not bring a row back.""" + day = self._seed_file(days_ago=3, status=ExecutionStatus.PENDING) + + # Still running: nothing to count yet. + _run_aggregation() + assert not EventMetricsDaily._base_manager.filter(date=day).exists() + + # It finishes. status turns terminal; created_at does not move. + WorkflowFileExecution.objects.update(status=ExecutionStatus.COMPLETED.value) + + # The per-run window no longer reaches its created_at, so it stays missed. + _run_aggregation() + assert not EventMetricsDaily._base_manager.filter(date=day).exists() + + # Only the wider pass recovers it. + _run_aggregation(source_window_days=DASHBOARD_RECONCILE_WINDOW_DAYS) + assert EventMetricsDaily._base_manager.filter( + date=day, metric_name="documents_processed" + ).exists() + + def test_gap_older_than_the_reconcile_window_needs_a_manual_backfill(self): + """Neither scheduled pass reaches a day beyond the reconcile window.""" + old_day = self._seed_file(days_ago=62) + recent_day = self._seed_file(days_ago=0) + + _run_aggregation() + _run_aggregation(source_window_days=DASHBOARD_RECONCILE_WINDOW_DAYS) + + # The run worked — it just cannot reach that far back. + assert EventMetricsDaily._base_manager.filter(date=recent_day).exists() + assert not EventMetricsDaily._base_manager.filter(date=old_day).exists() + + +class TestReconciliationSchedule(TestCase): + """Migration 0004 schedules the once-daily reconciliation pass. + + The suite runs with --no-migrations, so the migration's function is called + directly rather than relying on it having been applied. + """ + + def setUp(self): + """Load the data migration module.""" + self.migration = import_module( + "dashboard_metrics.migrations.0004_add_reconciliation_task" + ) + + def _task(self): + return PeriodicTask.objects.get(name=self.migration.RECONCILE_TASK_NAME) + + def test_migration_schedules_the_pass_at_0400_with_a_7_day_window(self): + """The beat row lands enabled, at 04:00 UTC, carrying the wider window.""" + self.migration.create_reconciliation_task(apps, None) + + task = self._task() + assert task.task == "dashboard_metrics.aggregate_from_sources" + assert task.enabled + assert task.queue == "dashboard_metric_events" + assert json.loads(task.kwargs) == { + "source_window_days": DASHBOARD_RECONCILE_WINDOW_DAYS + } + assert (task.crontab.hour, task.crontab.minute) == ("4", "0") + + def test_migration_is_idempotent_and_reversible(self): + """Re-running leaves one row; the reverse function removes it.""" + self.migration.create_reconciliation_task(apps, None) + self.migration.create_reconciliation_task(apps, None) + + assert ( + PeriodicTask.objects.filter( + name=self.migration.RECONCILE_TASK_NAME + ).count() + == 1 + ) + + self.migration.remove_reconciliation_task(apps, None) + assert not PeriodicTask.objects.filter( + name=self.migration.RECONCILE_TASK_NAME + ).exists() From 86ed04c5b0e770cf651cdae80d90287dc9d76fe5 Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Sat, 29 Aug 2026 13:00:06 +0530 Subject: [PATCH 4/7] UN-3973 Trim comments in tasks.py and revert the unrelated settings change MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cut the verbose comments and docstrings down to the purpose and the non-obvious bits. Code is unchanged. Restore backend/settings/base.py to main — moving the dotenv load ahead of get_required_setting was a local test convenience, not part of this change. The test rig exports the broker vars itself, so CI never needed it. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01KGZBF68CShem3pbUJM2tBc --- backend/backend/settings/base.py | 8 +-- backend/dashboard_metrics/tasks.py | 96 +++++++----------------------- 2 files changed, 26 insertions(+), 78 deletions(-) diff --git a/backend/backend/settings/base.py b/backend/backend/settings/base.py index f0dbafe6d2..d14f87b304 100644 --- a/backend/backend/settings/base.py +++ b/backend/backend/settings/base.py @@ -55,10 +55,6 @@ def get_required_setting(setting_key: str, default: str | None = None) -> str | # Load default log from env DEFAULT_LOG_LEVEL = os.environ.get("DEFAULT_LOG_LEVEL", "INFO") -ENV_FILE = find_dotenv() -if ENV_FILE: - load_dotenv(ENV_FILE) - # Celery Broker Configuration CELERY_BROKER_BASE_URL = get_required_setting("CELERY_BROKER_BASE_URL") CELERY_BROKER_USER = get_required_setting("CELERY_BROKER_USER") @@ -69,6 +65,10 @@ def get_required_setting(setting_key: str, default: str | None = None) -> str | ) ) +ENV_FILE = find_dotenv() +if ENV_FILE: + load_dotenv(ENV_FILE) + # Loading environment variables WORKFLOW_ACTION_EXPIRATION_TIME_IN_SECOND = os.environ.get( diff --git a/backend/dashboard_metrics/tasks.py b/backend/dashboard_metrics/tasks.py index d01b9b5418..c0b1b1a9b8 100644 --- a/backend/dashboard_metrics/tasks.py +++ b/backend/dashboard_metrics/tasks.py @@ -36,19 +36,15 @@ DASHBOARD_HOURLY_METRICS_RETENTION_DAYS = 30 DASHBOARD_DAILY_METRICS_RETENTION_DAYS = 365 -# Source lookback for the daily tier. Sized against the measured worst -# created_at -> terminal-status lag of ~2h, bounded by the file processing -# time limit and the stuck-execution reaper. +# Daily-tier source lookback, sized against the worst observed +# created_at -> terminal-status lag. DASHBOARD_SOURCE_WINDOW_DAYS = 2 -# Wider lookback used by the once-daily reconciliation pass, which repairs -# the daily tier after cron downtime. +# Wider lookback for the once-daily reconciliation pass. DASHBOARD_RECONCILE_WINDOW_DAYS = 7 -# Lookback for the active-org prefilter. Deliberately independent of the -# source window: metrics filtered on a column other than -# WorkflowExecution.created_at (e.g. hitl_completions on approved_at) can -# land for an org whose executions are older than the source window. +# Wider than the source window: metrics keyed on another column +# (e.g. approved_at) can land for an org whose executions are older. DASHBOARD_ACTIVE_ORG_LOOKBACK_DAYS = 7 @@ -184,18 +180,10 @@ def _bulk_upsert_daily(aggregations: dict) -> int: def _delete_orphan_monthly(month_start: date, fresh_keys: set[tuple]) -> int: - """Drop monthly rows in the window that the rollup no longer produces. + """Drop monthly rows from month_start that the rollup no longer produces. - Monthly is a pure derivation of the daily tier, so a key whose daily rows - have gone (a deleted workflow cascades to its executions) must not survive - as a stale total. - - Args: - month_start: First day of the earliest month being rebuilt - fresh_keys: Keys the current rollup produced - - Returns: - Number of rows deleted + Monthly derives from the daily tier, so a key with no daily rows left must + not survive as a stale total. """ stale_pks = [ row["pk"] @@ -219,22 +207,11 @@ def _delete_orphan_monthly(month_start: date, fresh_keys: set[tuple]) -> int: def _rollup_monthly_from_daily(month_start: date) -> int: - """Derive monthly metrics by summing the daily tier from month_start onwards. - - Replaces per-org monthly queries against source tables with a single - aggregate over event_metrics_daily, which retains far more history than - the monthly window needs. Rows in the window that the daily tier no longer - backs are removed, so the two tiers cannot disagree. + """Sum the daily tier from month_start into monthly, for all orgs at once. metric_type is aggregated rather than grouped: it is not part of unique_monthly_metric, so grouping on it could yield two rows for one conflict target. - - Args: - month_start: First day of the earliest month to rebuild - - Returns: - Number of rows upserted """ rows = ( EventMetricsDaily._base_manager.filter(date__gte=month_start) @@ -261,8 +238,8 @@ def _rollup_monthly_from_daily(month_start: date) -> int: for row in rows ] - # An empty daily tier means the source of truth is gone, not that every - # month is genuinely zero, so leave the existing rows alone. + # An empty daily tier means the source is gone, not that every month is + # zero — leave existing rows alone. if not objects: return 0 @@ -341,25 +318,14 @@ def _acquire_aggregation_lock() -> bool: def aggregate_metrics_from_sources( source_window_days: int = DASHBOARD_SOURCE_WINDOW_DAYS, ) -> dict[str, Any]: - """Aggregate metrics from source tables into hourly, daily, and monthly tables. - - This task runs periodically (every 15 minutes) to query metrics from - source tables (Usage, PageUsage, WorkflowExecution, etc.) and aggregate - them into EventMetricsHourly, EventMetricsDaily, and EventMetricsMonthly - tables for fast dashboard queries at different granularities. - - Uses a Redis distributed lock with self-healing to prevent overlapping - runs. If a previous run was killed without releasing the lock, the next - run detects the stale lock and reclaims it automatically. + """Aggregate source tables into the hourly, daily and monthly tiers. - Aggregation windows: - - Hourly: Last 24 hours (rolling window) - - Daily: source_window_days (covers late-arriving data) - - Monthly: Rolled up from the daily tier, current + previous month + 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. Args: source_window_days: Daily-tier source lookback. The once-daily - reconciliation pass runs the same task at + reconciliation pass reruns this task at DASHBOARD_RECONCILE_WINDOW_DAYS to repair gaps after downtime. Returns: @@ -387,11 +353,7 @@ def _aggregate_single_metric( daily_agg: dict, extra_kwargs: dict | None = None, ) -> None: - """Run a single metric query at hourly and daily granularity. - - Monthly totals are derived separately by rolling up the daily tier, so - neither query reaches further back than daily_start. - """ + """Run a single metric query at hourly and daily granularity.""" extra_kwargs = extra_kwargs or {} # === HOURLY (last 24h) === @@ -430,8 +392,7 @@ def _aggregate_llm_combined( ) -> None: """Run the combined LLM metrics query at hourly and daily granularity. - Issues 2 queries covering 4 metrics. Same windowing as - _aggregate_single_metric — monthly is derived from the daily tier. + Two queries covering four metrics. """ # === HOURLY (last 24h) === for row in MetricsQueryService.get_llm_metrics_combined( @@ -488,7 +449,7 @@ def _collect_org_metrics( daily_start: datetime, end_date: datetime, ) -> tuple[dict, dict, int]: - """Query every metric for one organization into hourly/daily aggregates. + """Query every metric for one org into hourly/daily aggregates. A failing metric is logged and counted, leaving the rest to proceed. @@ -502,7 +463,7 @@ def _collect_org_metrics( for metric_name, query_method, is_histogram in METRIC_CONFIGS: metric_type = MetricType.HISTOGRAM if is_histogram else MetricType.COUNTER - # Pre-resolved identifier spares PageUsage an Organization lookup per call. + # Pre-resolved identifier spares PageUsage a lookup per call. extra_kwargs = ( {"org_identifier": org.organization_id} if metric_name == "pages_processed" @@ -555,7 +516,6 @@ def _aggregate_org( ) stats["errors"] += errors - # Bulk upsert both tiers (single INSERT...ON CONFLICT each) if hourly_agg: stats["hourly"]["upserted"] += _bulk_upsert_hourly(hourly_agg) @@ -566,12 +526,7 @@ def _aggregate_org( def _active_org_ids(end_date: datetime) -> set: - """Organizations with recent execution activity. - - Deliberately wider than the source window: metrics filtered on a column - other than WorkflowExecution.created_at (hitl_completions uses approved_at) - can land for an org whose executions are older. - """ + """Organizations with execution activity in the prefilter lookback.""" return set( WorkflowExecution.objects.filter( created_at__gte=end_date - timedelta(days=DASHBOARD_ACTIVE_ORG_LOOKBACK_DAYS), @@ -611,16 +566,10 @@ def _build_result( def _run_aggregation( source_window_days: int = DASHBOARD_SOURCE_WINDOW_DAYS, ) -> dict[str, Any]: - """Execute the actual aggregation logic. - - Separated from the task function to keep the lock management clean. - """ + """Execute the aggregation, separately from the task's lock handling.""" end_date = timezone.now() - # Query windows for each granularity - # - Hourly: Last 24 hours (rolling window, matches retention of 30 days) - # - Daily: source_window_days of source data - # - Monthly: rolled up from the daily tier, current + previous month + # Monthly spans the current and previous month. hourly_start = end_date - timedelta(hours=24) daily_start = _truncate_to_day(end_date - timedelta(days=source_window_days)) monthly_start = _truncate_to_month( @@ -664,7 +613,6 @@ def _run_aggregation( logger.exception("Error processing org %s", org.id) stats["errors"] += 1 - # Monthly is derived from the daily tier in one statement for all orgs try: stats["monthly"]["upserted"] = _rollup_monthly_from_daily(monthly_start) except Exception: From 01e01f0873d9495a6452d1847bd7058759ba0762 Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Tue, 1 Sep 2026 19:20:50 +0530 Subject: [PATCH 5/7] UN-3973 Renumber the reconciliation migration to 0005 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit UN-3445 landed 0004_pg_periodic_tasks on main after this branch was cut, leaving dashboard_metrics with two 0004s depending on 0003 and nothing depending on either. Django saw two leaf nodes and refused to build the graph, so `migrate` failed before applying anything — every app, not just this one. Depend on 0004_pg_periodic_tasks and renumber to match, so the short prefix form stays usable for a rollback. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01KGZBF68CShem3pbUJM2tBc --- ...d_reconciliation_task.py => 0005_add_reconciliation_task.py} | 2 +- backend/dashboard_metrics/tests/test_tasks.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) rename backend/dashboard_metrics/migrations/{0004_add_reconciliation_task.py => 0005_add_reconciliation_task.py} (95%) diff --git a/backend/dashboard_metrics/migrations/0004_add_reconciliation_task.py b/backend/dashboard_metrics/migrations/0005_add_reconciliation_task.py similarity index 95% rename from backend/dashboard_metrics/migrations/0004_add_reconciliation_task.py rename to backend/dashboard_metrics/migrations/0005_add_reconciliation_task.py index 09667e867f..8a44759eed 100644 --- a/backend/dashboard_metrics/migrations/0004_add_reconciliation_task.py +++ b/backend/dashboard_metrics/migrations/0005_add_reconciliation_task.py @@ -49,7 +49,7 @@ def remove_reconciliation_task(apps, schema_editor): class Migration(migrations.Migration): dependencies = [ - ("dashboard_metrics", "0003_alter_eventmetricsdaily_organization_and_more"), + ("dashboard_metrics", "0004_pg_periodic_tasks"), ("django_celery_beat", "0018_improve_crontab_helptext"), ] diff --git a/backend/dashboard_metrics/tests/test_tasks.py b/backend/dashboard_metrics/tests/test_tasks.py index 6cb9316d1c..822c197a63 100644 --- a/backend/dashboard_metrics/tests/test_tasks.py +++ b/backend/dashboard_metrics/tests/test_tasks.py @@ -495,7 +495,7 @@ class TestReconciliationSchedule(TestCase): def setUp(self): """Load the data migration module.""" self.migration = import_module( - "dashboard_metrics.migrations.0004_add_reconciliation_task" + "dashboard_metrics.migrations.0005_add_reconciliation_task" ) def _task(self): From b974eaea5cdf8db30519eb19414167148e0ee374 Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Wed, 2 Sep 2026 16:49:15 +0530 Subject: [PATCH 6/7] UN-3973 [FIX] Address review: PG transport, scoped orphan sweep, retry posture MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reconciliation row could not run on the PG transport — two functions share the task name dashboard_metrics.aggregate_from_sources and the worker one took no arguments, so the mirrored row dispatched source_window_days into a zero-arg function and the message was dropped. The worker proxy and the internal endpoint now plumb it, and 0005 declares the PG twin rather than leaving the mirror to invent one. The orphan sweep is scoped to the (organization, month) partitions the rollup actually produced: an incomplete daily tier passed the empty-tier guard and deleted monthly rows it could not vouch for. Its deletion count now reaches the task result and a WARNING. DatabaseError and OperationalError propagate from the monthly rollup so the configured autoretry fires, instead of being logged once behind success: True. The prefilter is never narrower than the query window, so a widened source_window_days cannot skip the orgs it exists to repair. bulk_create takes an explicit batch_size. The Beat/PG drift guard named 0002 and 0004, so it kept comparing three schedules against three while this PR added a fourth. It now discovers every migration in the app, replays their RunPython forwards in order, derives the Beat cadence from the schedule row, binds every declared kwarg to its task signature, and asserts every post-install Beat write bumps PeriodicTasks.last_update. The reconcile row moves to 04:40 UTC: */15 fires at minute 0, and the per-tier lock keys do not block each other, so 04:00 started two full aggregations at once. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01KGZBF68CShem3pbUJM2tBc --- backend/dashboard_metrics/README.md | 28 +- backend/dashboard_metrics/internal_views.py | 13 +- .../management/commands/backfill_metrics.py | 25 +- .../0005_add_reconciliation_task.py | 99 ++++- backend/dashboard_metrics/tasks.py | 115 ++++-- .../test_pg_periodic_task_declarations.py | 279 +++++++++---- backend/dashboard_metrics/tests/test_tasks.py | 368 ++++++++++++++++-- workers/scheduler/dashboard_metrics_tasks.py | 17 +- workers/tests/test_dashboard_metrics_tasks.py | 27 ++ 9 files changed, 802 insertions(+), 169 deletions(-) diff --git a/backend/dashboard_metrics/README.md b/backend/dashboard_metrics/README.md index fc40e9f55b..d4f77b9eb7 100644 --- a/backend/dashboard_metrics/README.md +++ b/backend/dashboard_metrics/README.md @@ -46,7 +46,8 @@ celery -A backend beat -l info ### Celery Tasks & Schedule | Task | Schedule | What It Does | |------|----------|--------------| -| `aggregate_from_sources` | Every 15 min | Aggregates source → hourly/daily/monthly | +| `aggregate_from_sources` | Every 15 min | Aggregates source → hourly/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 | | `cleanup_hourly_data` | Daily 2 AM | Deletes hourly data > 30 days | | `cleanup_daily_data` | Weekly Sun 3 AM | Deletes daily data > 365 days | @@ -109,7 +110,7 @@ celery -A backend beat -l info │ EventMetrics │ │ EventMetrics │ │ EventMetrics │ │ Hourly │ │ Daily │ │ Monthly │ │ │ │ │ │ │ -│ • 24h query │ │ • 7 day query │ │ • 2 month query │ +│ • 24h query │ │ • 2 day query │ │ • from daily │ │ • 30 day retain │ │ • 365 day retain│ │ • No cleanup │ └────────┬────────┘ └────────┬────────┘ └────────┬────────┘ │ │ │ @@ -163,6 +164,7 @@ The dashboard reads from **pre-aggregated tables** (`event_metrics_hourly`, `eve **Failure Resilience:** - If the aggregation task fails, the dashboard shows stale data (up to 15 minutes old) rather than crashing. +- 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. @@ -298,8 +300,8 @@ cost = (input_cost_per_token × input_tokens) + (output_cost_per_token × output | Table | Model | Time Column | Granularity | Query Window | Retention | |-------|-------|-------------|-------------|--------------|-----------| | `event_metrics_hourly` | `EventMetricsHourly` | `timestamp` | Hour | Last 24 hours | 30 days | -| `event_metrics_daily` | `EventMetricsDaily` | `date` | Day | Last 7 days | 365 days | -| `event_metrics_monthly` | `EventMetricsMonthly` | `month` | Month | Last 2 months | Forever | +| `event_metrics_daily` | `EventMetricsDaily` | `date` | Day | Last 2 days (7 on the daily reconciliation pass) | 365 days | +| `event_metrics_monthly` | `EventMetricsMonthly` | `month` | Month | Rolled up from the daily tier, current + previous month | Forever | ### Table Schema @@ -340,6 +342,7 @@ 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` | | `cleanup_hourly_metrics` | `dashboard_metrics.cleanup_hourly_data` | Daily 2:00 AM UTC | `dashboard_metric_events` | Delete hourly data >30 days | | `cleanup_daily_metrics` | `dashboard_metrics.cleanup_daily_data` | Weekly Sun 3:00 AM UTC | `dashboard_metric_events` | Delete daily data >365 days | @@ -378,16 +381,23 @@ The `aggregate_metrics_from_sources` task: 2. **For each metric**: - Queries source table with `MetricsQueryService` - Groups by time period (hour/day/month) -3. **Upserts results** into aggregated tables using `update_or_create` -4. **Uses `_base_manager`** to bypass Django's organization filter in Celery context +3. **Upserts results** into the hourly and daily tables +4. **Rolls monthly up from the daily tier** in one statement for all orgs, dropping + monthly rows the daily tier no longer produces (scoped to the organization/month + partitions the rollup covered) +5. **Uses `_base_manager`** to bypass Django's organization filter in Celery context ```python # Query windows -hourly_start = end_date - timedelta(hours=24) # Last 24 hours -daily_start = end_date - timedelta(days=7) # Last 7 days -monthly_start = first_of_previous_month # Last 2 months +hourly_start = end_date - timedelta(hours=24) # Last 24 hours +daily_start = truncate_to_day(end_date - source_window_days) # 2 days, 7 on reconcile +monthly_start = first_of_previous_month # summed from daily ``` +The monthly tier has no source queries of its own. `backfill_metrics` still computes +monthly from source, so within the rollup window (current + previous month) its output +is recomputed within 15 minutes — see that command's help text. + --- ## API Endpoints diff --git a/backend/dashboard_metrics/internal_views.py b/backend/dashboard_metrics/internal_views.py index f776633944..0ecc6759f2 100644 --- a/backend/dashboard_metrics/internal_views.py +++ b/backend/dashboard_metrics/internal_views.py @@ -34,6 +34,7 @@ from utils.local_context import StateStore from dashboard_metrics.tasks import ( + DASHBOARD_SOURCE_WINDOW_DAYS, aggregate_metrics_from_sources, cleanup_daily_metrics, cleanup_hourly_metrics, @@ -94,7 +95,17 @@ class AggregateMetricsAPIView(_MetricsTaskAPIView): """ def post(self, request: Request) -> Response: - return self._run(aggregate_metrics_from_sources) + """``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) + try: + days = _int_arg(request, "source_window_days", DASHBOARD_SOURCE_WINDOW_DAYS) + except ValueError as exc: + return Response({"error": str(exc)}, status=status.HTTP_400_BAD_REQUEST) + return self._run(aggregate_metrics_from_sources, source_window_days=days) class CleanupHourlyMetricsAPIView(_MetricsTaskAPIView): diff --git a/backend/dashboard_metrics/management/commands/backfill_metrics.py b/backend/dashboard_metrics/management/commands/backfill_metrics.py index 9c4d82baca..d598e366bb 100644 --- a/backend/dashboard_metrics/management/commands/backfill_metrics.py +++ b/backend/dashboard_metrics/management/commands/backfill_metrics.py @@ -3,6 +3,12 @@ This command populates EventMetricsHourly, EventMetricsDaily, and EventMetricsMonthly tables from historical data in source tables (Usage, PageUsage, WorkflowExecution, etc.) +The current and previous month are owned exclusively by the aggregation task, which +derives them from the daily tier and drops monthly rows the daily tier no longer +produces. Inside that window this command's monthly output is recomputed within 15 +minutes, so --skip-monthly is a no-op and --skip-daily leaves monthly to be rebuilt +from a tier this run did not populate. Backfill both, or neither. + Usage: python manage.py backfill_metrics --days=30 python manage.py backfill_metrics --days=90 --org-id=5 @@ -92,12 +98,18 @@ def add_arguments(self, parser): parser.add_argument( "--skip-daily", action="store_true", - help="Skip daily aggregation", + help=( + "Skip daily aggregation. Unsafe for the current and previous month: " + "the aggregation task rebuilds monthly from daily there." + ), ) parser.add_argument( "--skip-monthly", action="store_true", - help="Skip monthly aggregation", + help=( + "Skip monthly aggregation. A no-op for the current and previous " + "month, which the aggregation task owns." + ), ) parser.add_argument( "--active-only", @@ -123,6 +135,15 @@ def handle(self, *args, **options): self.stdout.write(f"Backfill period: {start_date.date()} to {end_date.date()}") self.stdout.write(f"Days: {days}") + if skip_daily and not skip_monthly: + self.stdout.write( + self.style.WARNING( + "--skip-daily without --skip-monthly: the aggregation task " + "rebuilds the current and previous month from the daily tier " + "and will drop monthly rows this run writes there." + ) + ) + if dry_run: self.stdout.write(self.style.WARNING("DRY RUN - no changes will be made")) diff --git a/backend/dashboard_metrics/migrations/0005_add_reconciliation_task.py b/backend/dashboard_metrics/migrations/0005_add_reconciliation_task.py index 8a44759eed..d7bf134aab 100644 --- a/backend/dashboard_metrics/migrations/0005_add_reconciliation_task.py +++ b/backend/dashboard_metrics/migrations/0005_add_reconciliation_task.py @@ -3,21 +3,47 @@ The 15-minute aggregation reads a narrow source window, which cannot repair gaps left by cron downtime. This runs the same task once a day at a wider window to backfill them. + +Declared for **both** transports, like 0002/0004: Beat reads +``django_celery_beat_periodictask``, the PG scheduler reads ``pg_periodic_task``, +and a schedule present on one only stops firing the moment the flag flips. +``kwargs`` is a JSON string on Beat and a JSONField on PG — same value, two +encodings. """ from django.db import migrations +from django.utils import timezone RECONCILE_TASK_NAME = "dashboard_metrics_reconcile_source_window" +RECONCILE_DESCRIPTION = ( + "Re-aggregate metrics over a 7 day source window to repair " + "daily-tier gaps left by cron downtime" +) + +# Single source for both directions, and importable by the drift test. +PG_PERIODIC_TASKS = [ + { + "name": RECONCILE_TASK_NAME, + "task_name": "dashboard_metrics.aggregate_from_sources", + "queue": "dashboard_metric_events", + "task_args": [], + "task_kwargs": {"source_window_days": 7}, + # Beat: CrontabSchedule(minute=40, hour=4, every day) UTC — clear of the + # 2:00 and 3:00 cleanup tasks, and off the aggregation's */15 grid + # (:00 :15 :30 :45) so the two never start together. + "cron_string": "40 4 * * *", + }, +] def create_reconciliation_task(apps, schema_editor): - """Create the once-daily reconciliation periodic task.""" + """Create the once-daily reconciliation periodic task on both transports.""" crontab_model = apps.get_model("django_celery_beat", "CrontabSchedule") periodic_task_model = apps.get_model("django_celery_beat", "PeriodicTask") + pg_periodic_task_model = apps.get_model("pg_queue", "PgPeriodicTask") - # 4:00 AM UTC — clear of the 2:00 and 3:00 cleanup tasks schedule_4am, _ = crontab_model.objects.get_or_create( - minute="0", + minute="40", hour="4", day_of_week="*", day_of_month="*", @@ -25,32 +51,65 @@ def create_reconciliation_task(apps, schema_editor): defaults={"timezone": "UTC"}, ) - periodic_task_model.objects.update_or_create( - name=RECONCILE_TASK_NAME, - defaults={ - "task": "dashboard_metrics.aggregate_from_sources", - "crontab": schedule_4am, - "queue": "dashboard_metric_events", - "kwargs": '{"source_window_days": 7}', - "enabled": True, - "description": ( - "Re-aggregate metrics over a 7 day source window to repair " - "daily-tier gaps left by cron downtime" - ), - }, - ) + for spec in PG_PERIODIC_TASKS: + periodic_task_model.objects.update_or_create( + name=spec["name"], + defaults={ + "task": spec["task_name"], + "crontab": schedule_4am, + "queue": spec["queue"], + "kwargs": '{"source_window_days": 7}', + "enabled": True, + "description": RECONCILE_DESCRIPTION, + }, + ) + pg_periodic_task_model.objects.update_or_create( + name=spec["name"], + defaults={ + "task_name": spec["task_name"], + "queue": spec["queue"], + "task_args": spec["task_args"], + "task_kwargs": spec["task_kwargs"], + "cron_string": spec["cron_string"], + "org_id": "", + "enabled": True, + # Inert until the rollout flag decides otherwise. + "pg_owned": False, + }, + ) + + _bump_beat_change_tracker(apps) def remove_reconciliation_task(apps, schema_editor): - """Remove the reconciliation periodic task on rollback.""" - periodic_task_model = apps.get_model("django_celery_beat", "PeriodicTask") - periodic_task_model.objects.filter(name=RECONCILE_TASK_NAME).delete() + """Remove the reconciliation periodic task from both transports.""" + names = [spec["name"] for spec in PG_PERIODIC_TASKS] + apps.get_model("django_celery_beat", "PeriodicTask").objects.filter( + name__in=names + ).delete() + apps.get_model("pg_queue", "PgPeriodicTask").objects.filter(name__in=names).delete() + _bump_beat_change_tracker(apps) + + +def _bump_beat_change_tracker(apps): + """Make a running Beat reload instead of missing the new schedule. + + django-celery-beat's post_save receiver binds the concrete PeriodicTask, so + writes through a historical model never bump PeriodicTasks.last_update and + DatabaseScheduler keeps its stale in-memory copy. Same fix and reason as + scheduler/ownership.py and mirror_pg_periodic_tasks.py. + """ + periodic_tasks_model = apps.get_model("django_celery_beat", "PeriodicTasks") + periodic_tasks_model.objects.update_or_create( + ident=1, defaults={"last_update": timezone.now()} + ) class Migration(migrations.Migration): dependencies = [ ("dashboard_metrics", "0004_pg_periodic_tasks"), ("django_celery_beat", "0018_improve_crontab_helptext"), + ("pg_queue", "0003_pgperiodictask"), ] operations = [ diff --git a/backend/dashboard_metrics/tasks.py b/backend/dashboard_metrics/tasks.py index c0b1b1a9b8..332cb623ef 100644 --- a/backend/dashboard_metrics/tasks.py +++ b/backend/dashboard_metrics/tasks.py @@ -32,6 +32,10 @@ logger = logging.getLogger(__name__) +# Django 4.2's PostgreSQL backend does not override bulk_batch_size, so an +# unbatched bulk_create emits one statement whose size scales with tenant count. +MONTHLY_ROLLUP_BATCH_SIZE = 1000 + # Retention periods for metrics cleanup DASHBOARD_HOURLY_METRICS_RETENTION_DAYS = 30 DASHBOARD_DAILY_METRICS_RETENTION_DAYS = 365 @@ -40,7 +44,9 @@ # created_at -> terminal-status lag. DASHBOARD_SOURCE_WINDOW_DAYS = 2 -# Wider lookback for the once-daily reconciliation pass. +# Wider lookback for the once-daily reconciliation pass. A migration must not +# import live app code, so 0005_add_reconciliation_task carries this as a literal +# in the schedule row's kwargs — editing this constant does not move the schedule. DASHBOARD_RECONCILE_WINDOW_DAYS = 7 # Wider than the source window: metrics keyed on another column @@ -179,39 +185,54 @@ def _bulk_upsert_daily(aggregations: dict) -> int: return len(objects) -def _delete_orphan_monthly(month_start: date, fresh_keys: set[tuple]) -> int: - """Drop monthly rows from month_start that the rollup no longer produces. +def _delete_orphan_monthly(objects: list[EventMetricsMonthly]) -> int: + """Drop stale monthly rows inside the partitions the rollup actually covered. Monthly derives from the daily tier, so a key with no daily rows left must - not survive as a stale total. + not survive as a stale total. An (organization, month) the rollup produced + nothing for is left alone instead: absent daily rows there mean an + incomplete tier, not a metric that went to zero. """ - stale_pks = [ - row["pk"] - for row in EventMetricsMonthly._base_manager.filter( - month__gte=month_start - ).values("pk", "organization_id", "month", "metric_name", "project", "tag") - if ( - row["organization_id"], - row["month"], - row["metric_name"], - row["project"], - row["tag"], - ) - not in fresh_keys - ] - if not stale_pks: - return 0 + fresh_keys = { + (o.organization_id, o.month, o.metric_name, o.project, o.tag) for o in objects + } + covered: dict[date, set] = {} + for o in objects: + covered.setdefault(o.month, set()).add(o.organization_id) + + deleted = 0 + for month, org_ids in covered.items(): + stale_pks = [ + row["pk"] + for row in EventMetricsMonthly._base_manager.filter( + month=month, organization_id__in=org_ids + ).values("pk", "organization_id", "month", "metric_name", "project", "tag") + if ( + row["organization_id"], + row["month"], + row["metric_name"], + row["project"], + row["tag"], + ) + not in fresh_keys + ] + if not stale_pks: + continue + removed, _ = EventMetricsMonthly._base_manager.filter(pk__in=stale_pks).delete() + deleted += removed - deleted, _ = EventMetricsMonthly._base_manager.filter(pk__in=stale_pks).delete() return deleted -def _rollup_monthly_from_daily(month_start: date) -> int: +def _rollup_monthly_from_daily(month_start: date) -> tuple[int, int]: """Sum the daily tier from month_start into monthly, for all orgs at once. metric_type is aggregated rather than grouped: it is not part of unique_monthly_metric, so grouping on it could yield two rows for one conflict target. + + Returns: + (rows upserted, rows deleted as orphans) """ rows = ( EventMetricsDaily._base_manager.filter(date__gte=month_start) @@ -238,14 +259,11 @@ def _rollup_monthly_from_daily(month_start: date) -> int: for row in rows ] - # An empty daily tier means the source is gone, not that every month is - # zero — leave existing rows alone. + # Nothing to write and nothing covered, so nothing to sweep. The scoping in + # _delete_orphan_monthly already makes this safe; returning early just skips a + # pointless transaction. if not objects: - return 0 - - fresh_keys = { - (o.organization_id, o.month, o.metric_name, o.project, o.tag) for o in objects - } + return 0, 0 with transaction.atomic(): EventMetricsMonthly._base_manager.bulk_create( @@ -253,10 +271,11 @@ def _rollup_monthly_from_daily(month_start: date) -> int: update_conflicts=True, unique_fields=["organization", "month", "metric_name", "project", "tag"], update_fields=["metric_type", "metric_value", "metric_count"], + batch_size=MONTHLY_ROLLUP_BATCH_SIZE, ) - _delete_orphan_monthly(month_start, fresh_keys) + deleted = _delete_orphan_monthly(objects) - return len(objects) + return len(objects), deleted AGGREGATION_LOCK_KEY = "dashboard_metrics:aggregation_lock" @@ -525,12 +544,19 @@ def _aggregate_org( stats["orgs_processed"] += 1 -def _active_org_ids(end_date: datetime) -> set: - """Organizations with execution activity in the prefilter lookback.""" +def _active_org_ids(end_date: datetime, window_start: datetime) -> set: + """Organizations with execution activity in the prefilter lookback. + + Never narrower than the caller's own query window: a widened + source_window_days must not be prefiltered back down to the default + lookback, or the reconciliation pass skips the orgs it exists to repair. + """ + cutoff = min( + window_start, + end_date - timedelta(days=DASHBOARD_ACTIVE_ORG_LOOKBACK_DAYS), + ) return set( - WorkflowExecution.objects.filter( - created_at__gte=end_date - timedelta(days=DASHBOARD_ACTIVE_ORG_LOOKBACK_DAYS), - ) + WorkflowExecution.objects.filter(created_at__gte=cutoff) .values_list("workflow__organization_id", flat=True) .distinct() ) @@ -579,13 +605,13 @@ def _run_aggregation( stats = { "hourly": {"upserted": 0}, "daily": {"upserted": 0}, - "monthly": {"upserted": 0}, + "monthly": {"upserted": 0, "deleted": 0}, "errors": 0, "orgs_processed": 0, } # Pre-filter to orgs with recent activity to reduce DB load. - active_org_ids = _active_org_ids(end_date) + active_org_ids = _active_org_ids(end_date, daily_start) logger.info( "Aggregation: %d active orgs out of %d total", len(active_org_ids), @@ -614,7 +640,17 @@ def _run_aggregation( stats["errors"] += 1 try: - stats["monthly"]["upserted"] = _rollup_monthly_from_daily(monthly_start) + upserted, deleted = _rollup_monthly_from_daily(monthly_start) + stats["monthly"]["upserted"] = upserted + stats["monthly"]["deleted"] = deleted + if deleted: + logger.warning( + "Monthly rollup deleted %d orphan row(s) from %s", deleted, 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: logger.exception("Error rolling up monthly metrics from %s", monthly_start) stats["errors"] += 1 @@ -624,6 +660,7 @@ def _run_aggregation( f"hourly={stats['hourly']['upserted']}, " f"daily={stats['daily']['upserted']}, " f"monthly={stats['monthly']['upserted']}, " + f"monthly_deleted={stats['monthly']['deleted']}, " f"errors={stats['errors']}" ) 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 85ea407899..d83e2c2c69 100644 --- a/backend/dashboard_metrics/tests/test_pg_periodic_task_declarations.py +++ b/backend/dashboard_metrics/tests/test_pg_periodic_task_declarations.py @@ -1,113 +1,260 @@ """Drift guard between the Beat and PG declarations of the metrics periodics (UN-3796). -Two migrations declare the same three schedules — ``0002_setup_periodic_tasks`` for Celery -Beat and ``0004_pg_periodic_tasks`` for the PG scheduler. They are separate rows in -separate tables, so nothing stops someone editing one and forgetting the other. That is -the whole failure mode this file exists for: a schedule changed on Beat but not on PG means -the task silently runs on a different cadence the moment the flag flips. - -DB-free — both migration modules are imported and their declared specs compared directly, -so this runs in the unit tier rather than needing a migrated database. +Every schedule in this app is declared twice — once in +``django_celery_beat_periodictask`` for Celery Beat, once in ``pg_periodic_task`` for the +PG scheduler. They are separate rows in separate tables, so nothing stops someone editing +one and forgetting the other. That is the whole failure mode this file exists for: a +schedule changed on Beat but not on PG means the task silently runs on a different cadence +— or not at all — the moment the flag flips. + +**Every data migration in the app is replayed**, not a named pair. Naming modules is how +the guard went stale before: a schedule added in a later migration kept comparing the +original three against three and stayed green while the invariant it names was violated. +Migrations are run in order against fake models, so rows a later migration rewrites are +compared in their final state. + +DB-free — nothing here touches a database. """ from __future__ import annotations import importlib +import inspect import json +import re +from pathlib import Path +from types import SimpleNamespace import pytest +from django.db import migrations -_BEAT_MIGRATION = "dashboard_metrics.migrations.0002_setup_periodic_tasks" -_PG_MIGRATION = "dashboard_metrics.migrations.0004_pg_periodic_tasks" +from dashboard_metrics.tasks import ( + aggregate_metrics_from_sources, + cleanup_daily_metrics, + cleanup_hourly_metrics, +) -# Cron equivalent of each Beat schedule, asserted against what the Beat migration builds. -# Written out rather than derived: deriving it from the same code under test would make -# the comparison vacuous. +_MIGRATIONS_PKG = "dashboard_metrics.migrations" +_MIGRATIONS_DIR = Path(__file__).resolve().parent.parent / "migrations" + +# Cron equivalent of each schedule, written out rather than derived: an anchor that a +# reviewer reads, and that an edit to both declarations at once still has to touch. _EXPECTED_CRON = { "dashboard_metrics_aggregate_from_sources": "*/15 * * * *", "dashboard_metrics_cleanup_hourly": "0 2 * * *", "dashboard_metrics_cleanup_daily": "0 3 * * 0", + "dashboard_metrics_reconcile_source_window": "40 4 * * *", } -@pytest.fixture(scope="module") -def pg_specs() -> dict[str, dict]: - mod = importlib.import_module(_PG_MIGRATION) - return {spec["name"]: spec for spec in mod.PG_PERIODIC_TASKS} +class _Schedule: + """Stands in for an Interval/CrontabSchedule row, carrying its own cron string.""" + def __init__(self, **kwargs): + self.kwargs = kwargs -class _FakeQuerySet: - """Captures update_or_create calls from the Beat migration without a database.""" + @property + def cron_string(self) -> str: + k = self.kwargs + if "period" in k: + every, period = k["every"], k["period"] + if period == "minutes": + return f"*/{every} * * * *" + if period == "hours": + return f"0 */{every} * * *" + raise AssertionError(f"unhandled interval period: {period}") + return " ".join( + str(k[f]) + for f in ("minute", "hour", "day_of_month", "month_of_year", "day_of_week") + ) - def __init__(self, sink: dict): - self._sink = sink + +class _Rows: + """Captures a migration's writes to one model without a database.""" + + def __init__(self, factory=None): + self.rows: dict[str, dict] = {} + self.writes = 0 + self._factory = factory + self._selected: list[str] = [] def get_or_create(self, **kwargs): - # Schedule rows (Interval/Crontab) — return the kwargs so the PeriodicTask - # call can be inspected for which schedule it was given. - return kwargs, True + kwargs.pop("defaults", None) + return (self._factory(**kwargs) if self._factory else kwargs), True - def update_or_create(self, name=None, defaults=None, **_kw): - self._sink[name] = defaults or {} - return defaults, True + def update_or_create(self, name=None, defaults=None, **kwargs): + self.writes += 1 + if name is None: # e.g. PeriodicTasks(ident=1) — not a schedule row + return defaults, True + self.rows.setdefault(name, {}).update(defaults or {}) + return self.rows[name], True - def filter(self, *_a, **_k): + 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 update(self, **kwargs): + self.writes += 1 + for name in self._selected: + self.rows.setdefault(name, {}).update(kwargs) + return len(self._selected) + def delete(self): + self.writes += 1 + for name in self._selected: + self.rows.pop(name, None) return (0, {}) +class _Apps: + def __init__(self): + self.beat = _Rows() + self.pg = _Rows() + self.schedules = _Rows(factory=_Schedule) + self.tracker = _Rows() + self.other = _Rows() + + def get_model(self, app_label, model_name): + target = { + ("django_celery_beat", "PeriodicTask"): self.beat, + ("pg_queue", "PgPeriodicTask"): self.pg, + ("django_celery_beat", "CrontabSchedule"): self.schedules, + ("django_celery_beat", "IntervalSchedule"): self.schedules, + ("django_celery_beat", "PeriodicTasks"): self.tracker, + }.get((app_label, model_name), self.other) + return type("_M", (), {"objects": target}) + + +def _migration_modules() -> list[str]: + names = sorted( + p.stem for p in _MIGRATIONS_DIR.glob("*.py") if re.match(r"^\d{4}_", p.stem) + ) + assert names, "no migrations discovered — the glob is wrong, not the app" + return [f"{_MIGRATIONS_PKG}.{name}" for name in names] + + @pytest.fixture(scope="module") -def beat_specs() -> dict[str, dict]: - """Run the Beat migration's forward function against fakes and capture what it declares.""" - mod = importlib.import_module(_BEAT_MIGRATION) - captured: dict[str, dict] = {} +def declared() -> SimpleNamespace: + """Replay every data migration in order and capture what it declares.""" + apps = _Apps() + for dotted in _migration_modules(): + for op in importlib.import_module(dotted).Migration.operations: + if isinstance(op, migrations.RunPython): + op.code(apps, None) + return SimpleNamespace(beat=apps.beat.rows, pg=apps.pg.rows) - class _Apps: - def get_model(self, _app, model): - if model == "PeriodicTask": - return type("PT", (), {"objects": _FakeQuerySet(captured)}) - return type("S", (), {"objects": _FakeQuerySet({})}) - mod.create_periodic_tasks(_Apps(), None) - return captured +def _beat_cron(row: dict) -> str: + schedule = row.get("crontab") or row.get("interval") + assert schedule is not None, "Beat row declares neither a crontab nor an interval" + return schedule.cron_string class TestDeclarationsAgree: - def test_same_set_of_schedules(self, beat_specs, pg_specs): + def test_same_set_of_schedules(self, declared): # A schedule added to Beat but not PG stops firing the moment the flag flips; # the reverse fires something Beat never knew about. - assert set(beat_specs) == set(pg_specs) + assert set(declared.beat) == set(declared.pg) + + def test_every_known_schedule_is_declared(self, declared): + # Guards the guard: a replay that silently captured nothing would pass the + # set comparison above with two empty sets. + assert set(declared.beat) == set(_EXPECTED_CRON) - @pytest.mark.parametrize("name", sorted(_EXPECTED_CRON)) - def test_task_path_and_queue_match(self, beat_specs, pg_specs, name): - assert pg_specs[name]["task_name"] == beat_specs[name]["task"] - assert pg_specs[name]["queue"] == beat_specs[name]["queue"] + def test_task_path_and_queue_match(self, declared): + for name, beat in declared.beat.items(): + assert declared.pg[name]["task_name"] == beat["task"], name + assert declared.pg[name]["queue"] == beat["queue"], name - @pytest.mark.parametrize("name", sorted(_EXPECTED_CRON)) - def test_kwargs_match_once_decoded(self, beat_specs, pg_specs, name): + def test_kwargs_match_once_decoded(self, declared): # Beat stores kwargs as a JSON *string*; PgPeriodicTask.task_kwargs is a - # JSONField. A mismatch here means the cleanup runs with the wrong retention. - beat_kwargs = json.loads(beat_specs[name].get("kwargs") or "{}") - assert pg_specs[name]["task_kwargs"] == beat_kwargs + # JSONField. A mismatch means the task runs with the wrong arguments. + for name, beat in declared.beat.items(): + assert declared.pg[name]["task_kwargs"] == json.loads( + beat.get("kwargs") or "{}" + ), name + + def test_cadence_matches_across_transports(self, declared): + # Derived from the Beat schedule row rather than from a table, so a cadence + # changed on one transport only fails here whatever its name. + for name, beat in declared.beat.items(): + assert declared.pg[name]["cron_string"] == _beat_cron(beat), name - @pytest.mark.parametrize("name,cron", sorted(_EXPECTED_CRON.items())) - def test_cron_matches_the_beat_cadence(self, pg_specs, name, cron): - assert pg_specs[name]["cron_string"] == cron + def test_cadence_matches_the_written_anchor(self, declared): + for name, cron in _EXPECTED_CRON.items(): + assert declared.pg[name]["cron_string"] == cron + + +# 0002 seeds at install time, when Beat has never started and has nothing stale to +# reload. Every migration after it rewrites a schedule a running Beat already holds. +_INSTALL_MIGRATION = "0002_setup_periodic_tasks" + + +class TestRunningBeatIsToldToReload: + """Historical models fire no post_save, so DatabaseScheduler never reloads. + + Without an explicit ``PeriodicTasks.last_update`` bump a live Beat keeps firing its + in-memory copy: rows this migration adds never fire, rows it rewrites keep their old + arguments. Nothing errors, and the whole change silently does not happen. + """ + + def test_every_post_install_beat_write_bumps_the_change_tracker(self): + checked = 0 + for dotted in _migration_modules(): + if dotted.endswith(_INSTALL_MIGRATION): + continue + for op in importlib.import_module(dotted).Migration.operations: + if not isinstance(op, migrations.RunPython): + continue + for direction in (op.code, op.reverse_code): + if direction is None: + continue + apps = _Apps() + direction(apps, None) + if not apps.beat.writes: + continue + checked += 1 + assert apps.tracker.writes, f"{dotted}.{direction.__name__}" + assert checked, "no post-install Beat writes found — the discovery is broken" + + +class TestDeclaredKwargsAreCallable: + """A schedule row carrying a kwarg its task cannot bind raises TypeError per tick. + + TypeError is not in ``autoretry_for``, and the PG leg drops the message at + MAX_ATTEMPTS=1 — so the schedule silently never runs. Enumerating every declared + row rather than one migration's own spec is the point: the rows are added by + different migrations, and each new one is exactly the case that escapes a guard + scoped to a single module. + """ + + _TASKS = { + task.name: task + for task in ( + aggregate_metrics_from_sources, + cleanup_hourly_metrics, + cleanup_daily_metrics, + ) + } + + def test_every_declared_kwarg_set_binds_to_the_task_signature(self, declared): + for name, row in declared.pg.items(): + task = self._TASKS.get(row["task_name"]) + assert task is not None, f"{name} schedules an unknown task" + inspect.signature(task).bind(**row["task_kwargs"]) class TestSeededInert: - """Applying the migration must not cause anything to fire.""" - - def test_no_spec_declares_itself_pg_owned(self, pg_specs): - # pg_owned is set to False in the migration's defaults, never from the spec — - # this pins that no spec can smuggle ownership in. - assert not any("pg_owned" in spec for spec in pg_specs.values()) - - def test_no_spec_presets_a_run_time(self, pg_specs): - # A non-NULL next_run_at in the past would read as "overdue" and fire a burst - # of catch-up runs the moment the flag is enabled. - for spec in pg_specs.values(): - assert "next_run_at" not in spec - assert "last_run_at" not in spec + """Applying the migrations must not cause anything to fire.""" + + def test_nothing_is_declared_pg_owned(self, declared): + # pg_owned=True would hand the row to the PG scheduler before the rollout + # flag decides, and disable its Beat twin. + assert not any(row.get("pg_owned") for row in declared.pg.values()) + + def test_no_row_presets_a_run_time(self, declared): + # A non-NULL next_run_at in the past reads as "overdue" and fires a burst of + # catch-up runs the moment the flag is enabled. + for row in declared.pg.values(): + assert "next_run_at" not in row + assert "last_run_at" not in row diff --git a/backend/dashboard_metrics/tests/test_tasks.py b/backend/dashboard_metrics/tests/test_tasks.py index 822c197a63..e85acd203b 100644 --- a/backend/dashboard_metrics/tests/test_tasks.py +++ b/backend/dashboard_metrics/tests/test_tasks.py @@ -3,14 +3,16 @@ import json from datetime import date, datetime, timedelta from importlib import import_module +from types import SimpleNamespace from unittest.mock import patch from django.apps import apps from django.db import connection +from django.db.utils import DatabaseError from django.test import TestCase from django.test.utils import CaptureQueriesContext from django.utils import timezone -from django_celery_beat.models import PeriodicTask +from django_celery_beat.models import PeriodicTask, PeriodicTasks from account_v2.models import Organization from dashboard_metrics.models import ( @@ -19,13 +21,16 @@ EventMetricsMonthly, MetricType, ) +from pg_queue.models import PgPeriodicTask from workflow_manager.file_execution.models import WorkflowFileExecution from workflow_manager.workflow_v2.enums import ExecutionStatus from workflow_manager.workflow_v2.models.execution import WorkflowExecution from workflow_manager.workflow_v2.models.workflow import Workflow +from dashboard_metrics.internal_views import AggregateMetricsAPIView from dashboard_metrics.tasks import ( DASHBOARD_RECONCILE_WINDOW_DAYS, DASHBOARD_SOURCE_WINDOW_DAYS, + _active_org_ids, _rollup_monthly_from_daily, _run_aggregation, _truncate_to_day, @@ -226,12 +231,20 @@ def setUp(self): organization_id="rollup-org", name="rollup-org", display_name="Rollup Org" ) - def _daily(self, day, value, count=1, metric_type=MetricType.COUNTER): - """Create a daily metric row for the fixture org.""" + def _daily( + self, + day, + value, + count=1, + metric_type=MetricType.COUNTER, + metric_name="documents_processed", + org=None, + ): + """Create a daily metric row, defaulting to the fixture org and metric.""" EventMetricsDaily.objects.create( - organization=self.org, + organization=org or self.org, date=day, - metric_name="documents_processed", + metric_name=metric_name, metric_type=metric_type, metric_value=value, metric_count=count, @@ -239,15 +252,19 @@ def _daily(self, day, value, count=1, metric_type=MetricType.COUNTER): ) def _monthly_rows(self): - """Read back monthly rows ordered by month.""" - return list(EventMetricsMonthly._base_manager.order_by("month")) + """Read back monthly rows in a stable order.""" + return list( + EventMetricsMonthly._base_manager.order_by( + "month", "organization_id", "metric_name" + ) + ) def test_sums_daily_rows_into_month_bucket(self): """Daily rows within a month sum into a single monthly row.""" self._daily(date(2024, 3, 5), value=10, count=2) self._daily(date(2024, 3, 18), value=32, count=4) - assert _rollup_monthly_from_daily(date(2024, 3, 1)) == 1 + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (1, 0) rows = self._monthly_rows() assert len(rows) == 1 @@ -262,7 +279,7 @@ def test_month_boundary_keeps_months_separate(self): self._daily(date(2024, 2, 1), value=100) self._daily(date(2024, 2, 2), value=200) - assert _rollup_monthly_from_daily(date(2024, 1, 1)) == 2 + assert _rollup_monthly_from_daily(date(2024, 1, 1)) == (2, 0) rows = self._monthly_rows() assert [r.month for r in rows] == [date(2024, 1, 1), date(2024, 2, 1)] @@ -273,7 +290,7 @@ def test_excludes_months_before_the_window(self): self._daily(date(2023, 12, 15), value=999) self._daily(date(2024, 1, 15), value=5) - assert _rollup_monthly_from_daily(date(2024, 1, 1)) == 1 + assert _rollup_monthly_from_daily(date(2024, 1, 1)) == (1, 0) rows = self._monthly_rows() assert len(rows) == 1 @@ -297,30 +314,104 @@ def test_mixed_metric_type_within_a_month_yields_one_row(self): self._daily(date(2024, 3, 5), value=10, metric_type=MetricType.HISTOGRAM) self._daily(date(2024, 3, 6), value=5, metric_type=MetricType.COUNTER) - assert _rollup_monthly_from_daily(date(2024, 3, 1)) == 1 + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (1, 0) rows = self._monthly_rows() assert len(rows) == 1 assert rows[0].metric_value == 15 - def test_no_daily_rows_upserts_nothing(self): - """An empty daily tier is a no-op, not an error.""" - assert _rollup_monthly_from_daily(date(2024, 3, 1)) == 0 - assert not self._monthly_rows() + def test_an_empty_daily_tier_leaves_existing_monthly_rows_alone(self): + """An empty tier means the source is gone, not that every month is zero. - def test_monthly_row_is_dropped_once_its_daily_rows_are_gone(self): - """A month whose daily rows were deleted must not keep a stale total.""" + Seeding a monthly row first is what makes the failure reachable at all: with + an empty table a sweep that deletes everything and one that deletes nothing + both leave an empty table, and the assertion passes either way. + """ + EventMetricsMonthly._base_manager.create( + organization=self.org, + month=date(2024, 3, 1), + metric_name="documents_processed", + metric_type=MetricType.COUNTER, + metric_value=42, + metric_count=6, + project="default", + ) + + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (0, 0) + + rows = self._monthly_rows() + assert len(rows) == 1 + assert rows[0].metric_value == 42 + + def test_metric_is_dropped_once_its_daily_rows_are_gone(self): + """A metric with no daily rows left must not keep a stale monthly total.""" + self._daily(date(2024, 3, 5), value=10) + self._daily(date(2024, 3, 6), value=7, metric_name="pages_processed") + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (2, 0) + + EventMetricsDaily._base_manager.filter(metric_name="pages_processed").delete() + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (1, 1) + + rows = self._monthly_rows() + assert [r.metric_name for r in rows] == ["documents_processed"] + + def test_a_month_with_no_daily_rows_of_its_own_is_left_alone(self): + """An incomplete daily tier must read as "unknown", not as "deleted". + + A partially populated tier passes the empty-tier guard, so without scoping the + sweep to the (organization, month) partitions the rollup actually produced, + one missing month wipes that month's monthly rows for every org. + """ self._daily(date(2024, 3, 5), value=10) self._daily(date(2024, 4, 5), value=7) - _rollup_monthly_from_daily(date(2024, 3, 1)) - assert len(self._monthly_rows()) == 2 + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (2, 0) EventMetricsDaily._base_manager.filter(date=date(2024, 3, 5)).delete() - _rollup_monthly_from_daily(date(2024, 3, 1)) + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (1, 0) + + months = [row.month for row in self._monthly_rows()] + assert months == [date(2024, 3, 1), date(2024, 4, 1)] + + def test_the_sweep_is_scoped_per_organization(self): + """The sweep bypasses the org-scoped manager, so the org half of the key is + load-bearing: drop it and one org's daily rows vouch for another's monthly.""" + other = Organization.objects.create( + organization_id="rollup-org-2", name="rollup-org-2", display_name="Other" + ) + self._daily(date(2024, 3, 5), value=10) + self._daily(date(2024, 3, 6), value=20, org=other) + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (2, 0) + + EventMetricsDaily._base_manager.filter(organization=other).delete() + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (1, 0) rows = self._monthly_rows() - assert len(rows) == 1 - assert rows[0].month == date(2024, 4, 1) + assert [(r.organization_id, r.metric_value) for r in rows] == [ + (self.org.id, 10), + (other.id, 20), + ] + + def test_a_stale_row_is_dropped_for_one_organization_only(self): + """Two orgs in one month: only the org whose metric vanished loses its row.""" + other = Organization.objects.create( + organization_id="rollup-org-3", name="rollup-org-3", display_name="Other" + ) + self._daily(date(2024, 3, 5), value=10) + self._daily(date(2024, 3, 5), value=1, metric_name="pages_processed") + self._daily(date(2024, 3, 6), value=20, org=other) + self._daily(date(2024, 3, 6), value=2, metric_name="pages_processed", org=other) + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (4, 0) + + EventMetricsDaily._base_manager.filter( + organization=other, metric_name="pages_processed" + ).delete() + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (3, 1) + + survivors = { + (r.organization_id, r.metric_name) for r in self._monthly_rows() + } + assert (self.org.id, "pages_processed") in survivors + assert (other.id, "pages_processed") not in survivors def test_months_before_the_window_are_left_alone(self): """Orphan cleanup must not reach outside the rebuilt window.""" @@ -366,6 +457,181 @@ def test_monthly_rollup_never_touches_source_tables(self): assert source_table not in sql, f"monthly rollup read {source_table}" +class TestActiveOrgPrefilter(TestCase): + """The prefilter must never be narrower than the window it is filtering for.""" + + def setUp(self): + self.org = Organization.objects.create( + organization_id="prefilter-org", name="prefilter", display_name="Prefilter" + ) + workflow = Workflow.objects.create( + workflow_name="prefilter-wf", organization=self.org + ) + self.now = timezone.now() + execution = WorkflowExecution.objects.create( + workflow_id=workflow.id, status=ExecutionStatus.COMPLETED + ) + WorkflowExecution.objects.filter(pk=execution.pk).update( + created_at=self.now - timedelta(days=10) + ) + + def test_an_org_outside_the_default_lookback_is_filtered_out(self): + """The default lookback is the cheap case and stays exactly as wide as before.""" + window_start = self.now - timedelta(days=DASHBOARD_SOURCE_WINDOW_DAYS) + assert self.org.id not in _active_org_ids(self.now, window_start) + + def test_a_widened_window_widens_the_prefilter_with_it(self): + """Otherwise a long-outage repair queries 30 days for orgs active in 7, and + reports errors: 0 having skipped every org it exists to repair.""" + window_start = self.now - timedelta(days=30) + assert self.org.id in _active_org_ids(self.now, window_start) + + +class TestMonthlyRollupFailurePosture(TestCase): + """The rollup's errors must reach the task's retry, not a stats counter.""" + + def test_a_database_error_propagates_instead_of_reporting_success(self): + """DatabaseError/OperationalError are what autoretry_for is configured for. + + Swallowed here they become one INFO line and success: True, and a persistent + fault leaves monthly permanently stale behind 96 clean-looking runs a day. + """ + with patch("dashboard_metrics.tasks.WorkflowExecution") as mock_execution: + prefilter = mock_execution.objects.filter.return_value + prefilter.values_list.return_value.distinct.return_value = [1] + with patch( + "dashboard_metrics.tasks._rollup_monthly_from_daily", + side_effect=DatabaseError("lock timeout"), + ): + with self.assertRaises(DatabaseError): + _run_aggregation() + + def test_an_unexpected_error_is_still_counted_rather_than_fatal(self): + """Everything outside the retry set keeps the previous non-fatal posture.""" + with patch("dashboard_metrics.tasks.WorkflowExecution") as mock_execution: + prefilter = mock_execution.objects.filter.return_value + prefilter.values_list.return_value.distinct.return_value = [1] + with patch( + "dashboard_metrics.tasks._rollup_monthly_from_daily", + side_effect=ValueError("bad row"), + ): + result = _run_aggregation() + assert result["success"] is True + assert result["errors"] == 1 + + +class TestInternalAggregateEndpoint(TestCase): + """The PG transport reaches the task through this view, not through Celery.""" + + def _post(self, data): + return AggregateMetricsAPIView().post(SimpleNamespace(data=data)) + + def test_the_source_window_reaches_the_task(self): + with patch( + "dashboard_metrics.internal_views.aggregate_metrics_from_sources", + return_value={"success": True}, + ) as task: + self._post({"source_window_days": 7}) + assert task.call_args.kwargs == {"source_window_days": 7} + + def test_omitting_it_leaves_the_task_default_in_charge(self): + with patch( + "dashboard_metrics.internal_views.aggregate_metrics_from_sources", + return_value={"success": True}, + ) as task: + self._post({}) + assert task.call_args.kwargs == {} + + def test_a_non_integer_window_is_a_400(self): + with patch( + "dashboard_metrics.internal_views.aggregate_metrics_from_sources" + ) as task: + response = self._post({"source_window_days": "seven"}) + assert response.status_code == 400 + task.assert_not_called() + + +class TestMonthlyThroughTheTask(TestCase): + """The rollup as the task actually runs it, not via the helper directly. + + Every other rollup test calls ``_rollup_monthly_from_daily`` with a hand-chosen + ``month_start``. Nothing exercised the arithmetic that computes it, nor the sweep + running against a monthly table that already holds rows from earlier runs — so a + regression to "first of the current month" would silently drop last month's rows + with the whole rollup suite still green. + """ + + def setUp(self): + self.org = Organization.objects.create( + organization_id="entry-org", name="entry-org", display_name="Entry Org" + ) + now = timezone.now() + self.this_month = _truncate_to_month(now).date() + self.last_month = _truncate_to_month( + _truncate_to_month(now) - timedelta(days=1) + ).date() + self.before_window = _truncate_to_month( + _truncate_to_month(now - timedelta(days=1)) - timedelta(days=40) + ).date() + + def _daily(self, day, value, metric_name="documents_processed"): + EventMetricsDaily._base_manager.create( + organization=self.org, + date=day, + metric_name=metric_name, + metric_type=MetricType.COUNTER, + metric_value=value, + metric_count=1, + project="default", + tag="", + ) + + def _monthly(self, month, value, metric_name="documents_processed"): + EventMetricsMonthly._base_manager.create( + organization=self.org, + month=month, + metric_name=metric_name, + metric_type=MetricType.COUNTER, + metric_value=value, + metric_count=1, + project="default", + tag="", + ) + + def _run(self, **kwargs): + with patch("dashboard_metrics.tasks.WorkflowExecution") as mock_execution: + prefilter = mock_execution.objects.filter.return_value + prefilter.values_list.return_value.distinct.return_value = [self.org.id] + return _run_aggregation(**kwargs) + + def test_the_window_covers_the_previous_month_and_spares_what_precedes_it(self): + """monthly_start is the first of the *previous* month, and the sweep stops there.""" + self._daily(self.this_month, value=10) + self._daily(self.last_month, value=20) + self._monthly(self.before_window, value=999) + + result = self._run() + + assert result["period"]["monthly"]["start"] == self.last_month.isoformat() + assert result["monthly"] == {"upserted": 2, "deleted": 0} + + rows = EventMetricsMonthly._base_manager.order_by("month") + assert [r.month for r in rows] == [ + self.before_window, + self.last_month, + self.this_month, + ] + + def test_the_run_reports_what_it_deleted(self): + """The only destructive write in the task has to reach the result.""" + self._daily(self.this_month, value=10) + self._daily(self.this_month, value=1, metric_name="pages_processed") + assert self._run()["monthly"] == {"upserted": 2, "deleted": 0} + + EventMetricsDaily._base_manager.filter(metric_name="pages_processed").delete() + assert self._run()["monthly"] == {"upserted": 1, "deleted": 1} + + class TestSourceWindow(TestCase): """Tests for the per-run source window and the reconciliation pass.""" @@ -403,12 +669,25 @@ def test_reconciliation_window_widens_the_daily_query(self): assert result["period"]["daily"]["start"] == expected.isoformat() def test_task_passes_the_window_through(self): - """The scheduled task forwards its kwarg, defaulting to the per-run window.""" - with patch("dashboard_metrics.tasks._run_aggregation") as mock_run: + """The scheduled task forwards its kwarg, defaulting to the per-run window. + + The lock is patched out: acquiring it for real takes — and then releases in the + task's ``finally`` — the shared Redis key a live local aggregation may be + holding. + """ + with ( + patch("dashboard_metrics.tasks._acquire_aggregation_lock", return_value=True), + patch("dashboard_metrics.tasks.cache"), + patch("dashboard_metrics.tasks._run_aggregation") as mock_run, + ): aggregate_metrics_from_sources() mock_run.assert_called_once_with(DASHBOARD_SOURCE_WINDOW_DAYS) - with patch("dashboard_metrics.tasks._run_aggregation") as mock_run: + with ( + patch("dashboard_metrics.tasks._acquire_aggregation_lock", return_value=True), + patch("dashboard_metrics.tasks.cache"), + patch("dashboard_metrics.tasks._run_aggregation") as mock_run, + ): aggregate_metrics_from_sources(source_window_days=7) mock_run.assert_called_once_with(7) @@ -486,7 +765,7 @@ def test_gap_older_than_the_reconcile_window_needs_a_manual_backfill(self): class TestReconciliationSchedule(TestCase): - """Migration 0004 schedules the once-daily reconciliation pass. + """Migration 0005 schedules the once-daily reconciliation pass on both transports. The suite runs with --no-migrations, so the migration's function is called directly rather than relying on it having been applied. @@ -501,8 +780,8 @@ def setUp(self): def _task(self): return PeriodicTask.objects.get(name=self.migration.RECONCILE_TASK_NAME) - def test_migration_schedules_the_pass_at_0400_with_a_7_day_window(self): - """The beat row lands enabled, at 04:00 UTC, carrying the wider window.""" + def test_migration_schedules_the_pass_at_0440_with_a_7_day_window(self): + """The beat row lands enabled, at 04:40 UTC, carrying the wider window.""" self.migration.create_reconciliation_task(apps, None) task = self._task() @@ -512,7 +791,35 @@ def test_migration_schedules_the_pass_at_0400_with_a_7_day_window(self): assert json.loads(task.kwargs) == { "source_window_days": DASHBOARD_RECONCILE_WINDOW_DAYS } - assert (task.crontab.hour, task.crontab.minute) == ("4", "0") + assert (task.crontab.hour, task.crontab.minute) == ("4", "40") + + def test_the_pg_twin_lands_with_the_same_cadence_and_kwargs(self): + """A Beat-only row stops firing the moment the PG scheduler takes over.""" + self.migration.create_reconciliation_task(apps, None) + + row = PgPeriodicTask.objects.get(name=self.migration.RECONCILE_TASK_NAME) + assert row.task_name == "dashboard_metrics.aggregate_from_sources" + assert row.queue == "dashboard_metric_events" + assert row.cron_string == "40 4 * * *" + assert row.task_kwargs == { + "source_window_days": DASHBOARD_RECONCILE_WINDOW_DAYS + } + assert row.enabled + # Inert until the rollout flag decides otherwise. + assert not row.pg_owned + assert row.next_run_at is None + + def test_a_running_beat_is_told_to_reload(self): + """Historical models fire no post_save, so the tracker has to be bumped by hand. + + Without it a live Beat never adopts the new schedule and the reconciliation + pass simply never runs — no error, nothing logged. + """ + before = timezone.now() + self.migration.create_reconciliation_task(apps, None) + + tracker = PeriodicTasks.objects.get(ident=1) + assert tracker.last_update >= before def test_migration_is_idempotent_and_reversible(self): """Re-running leaves one row; the reverse function removes it.""" @@ -530,3 +837,6 @@ def test_migration_is_idempotent_and_reversible(self): assert not PeriodicTask.objects.filter( name=self.migration.RECONCILE_TASK_NAME ).exists() + assert not PgPeriodicTask.objects.filter( + name=self.migration.RECONCILE_TASK_NAME + ).exists() diff --git a/workers/scheduler/dashboard_metrics_tasks.py b/workers/scheduler/dashboard_metrics_tasks.py index 44bbe50440..07787aa9f3 100644 --- a/workers/scheduler/dashboard_metrics_tasks.py +++ b/workers/scheduler/dashboard_metrics_tasks.py @@ -105,9 +105,20 @@ def _log_if_skipped(name: str, result: dict[str, Any]) -> None: @worker_task(name="dashboard_metrics.aggregate_from_sources") -def dashboard_metrics_aggregate() -> dict[str, Any]: - """Aggregate source tables into the hourly/daily/monthly metrics tables.""" - result = _call_internal(_AGGREGATE_PATH) +def dashboard_metrics_aggregate( + 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. + """ + body = ( + {"source_window_days": source_window_days} + if source_window_days is not None + else 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 ce8ff853ac..b78cbedb01 100644 --- a/workers/tests/test_dashboard_metrics_tasks.py +++ b/workers/tests/test_dashboard_metrics_tasks.py @@ -8,6 +8,7 @@ from __future__ import annotations +import inspect import sys from pathlib import Path from unittest.mock import MagicMock, patch @@ -61,12 +62,38 @@ 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): with patch.object(dmt, "_call_internal", return_value={"success": True}) as call: 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) + + 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. + 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. + with patch.object(dmt, "_call_internal", return_value={"success": True}) as call: + dmt.dashboard_metrics_aggregate() + assert call.call_args.kwargs["body"] is None + @pytest.mark.parametrize( "func,path", [ From 2398a4332e5c57a9b45da1d03cc7f64dfe059bc7 Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Wed, 2 Sep 2026 17:26:38 +0530 Subject: [PATCH 7/7] UN-3973 [FIX] Address Athul's review: upsert-only monthly, per-schedule lock, honest success MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The orphan delete is removed. The design agreed on this ticket (comments 44768/45016) is a pure INSERT ... ON CONFLICT DO UPDATE; the delete was scope beyond it, and it converts a recoverable undercount into unrecoverable loss — the daily rows that would rebuild a deleted monthly row are exactly the ones that were missing. A stale total is recoverable with backfill_metrics. The reconciliation pass no longer shares a lock key with the 15-minute schedule. The 15-minute row is an IntervalSchedule and drifts against a fixed crontab, so on a shared key the once-daily repair loses the race roughly one day in seven, returns skipped=True and is never retried. A run in which every metric for every org failed no longer reports success: True. The result's success now reflects the error count, the completion log rises to WARNING, and the worker-side guard reads skipped_reason and errors as well as skipped — it saw none of these three did-nothing shapes before. A failed monthly rollup is distinguishable from an empty one: upserted=0 collided with the legitimate no-op and the no_active_orgs return. source_window_days is validated and bounded. It arrives as JSON from a Beat row that is editable in the admin: negative puts the window in the future, 0 never refreshes yesterday, 365 restores the multi-month scan this ticket exists to remove. Tests: a golden test seeds source rows, lets the real aggregation populate daily, and compares the rolled-up monthly against the pre-change derivation computed independently from get_documents_processed — AC-4 was claimed Met and had no equivalence assertion. Fixture offsets derive from the month boundary rather than fixed day counts, which land in the wrong month for the last days of any month. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01KGZBF68CShem3pbUJM2tBc --- backend/dashboard_metrics/README.md | 11 +- .../management/commands/backfill_metrics.py | 17 +- backend/dashboard_metrics/tasks.py | 176 +++++----- backend/dashboard_metrics/tests/test_tasks.py | 311 ++++++++++++++---- workers/scheduler/dashboard_metrics_tasks.py | 19 +- 5 files changed, 373 insertions(+), 161 deletions(-) diff --git a/backend/dashboard_metrics/README.md b/backend/dashboard_metrics/README.md index d4f77b9eb7..8c3fee6ce8 100644 --- a/backend/dashboard_metrics/README.md +++ b/backend/dashboard_metrics/README.md @@ -382,9 +382,10 @@ The `aggregate_metrics_from_sources` task: - Queries source table with `MetricsQueryService` - Groups by time period (hour/day/month) 3. **Upserts results** into the hourly and daily tables -4. **Rolls monthly up from the daily tier** in one statement for all orgs, dropping - monthly rows the daily tier no longer produces (scoped to the organization/month - partitions the rollup covered) +4. **Rolls monthly up from the daily tier** in one statement for all orgs. Upsert-only: + a monthly row the daily tier no longer produces is left in place. A stale total is + recoverable with `backfill_metrics`; a deleted one is not, because the daily rows + that would rebuild it are exactly what is missing 5. **Uses `_base_manager`** to bypass Django's organization filter in Celery context ```python @@ -396,7 +397,9 @@ monthly_start = first_of_previous_month # summed from dail The monthly tier has no source queries of its own. `backfill_metrics` still computes monthly from source, so within the rollup window (current + previous month) its output -is recomputed within 15 minutes — see that command's help text. +is overwritten within 15 minutes by the sum of the daily tier — see that command's help +text. **Backfill daily before relying on monthly:** the rollup writes whatever daily +holds, so a month whose daily tier is short produces an under-counted monthly total. --- diff --git a/backend/dashboard_metrics/management/commands/backfill_metrics.py b/backend/dashboard_metrics/management/commands/backfill_metrics.py index d598e366bb..a3af666ede 100644 --- a/backend/dashboard_metrics/management/commands/backfill_metrics.py +++ b/backend/dashboard_metrics/management/commands/backfill_metrics.py @@ -3,11 +3,11 @@ This command populates EventMetricsHourly, EventMetricsDaily, and EventMetricsMonthly tables from historical data in source tables (Usage, PageUsage, WorkflowExecution, etc.) -The current and previous month are owned exclusively by the aggregation task, which -derives them from the daily tier and drops monthly rows the daily tier no longer -produces. Inside that window this command's monthly output is recomputed within 15 -minutes, so --skip-monthly is a no-op and --skip-daily leaves monthly to be rebuilt -from a tier this run did not populate. Backfill both, or neither. +The current and previous month are recomputed from the daily tier by the aggregation +task every 15 minutes, so inside that window this command's monthly output is +overwritten and --skip-monthly is a no-op. --skip-daily is worse than useless there: +monthly is rebuilt from a tier this run did not populate, producing an under-count. +Backfill both, or neither. Usage: python manage.py backfill_metrics --days=30 @@ -100,7 +100,8 @@ def add_arguments(self, parser): action="store_true", help=( "Skip daily aggregation. Unsafe for the current and previous month: " - "the aggregation task rebuilds monthly from daily there." + "the aggregation task rebuilds monthly from daily there, so monthly " + "ends up under-counted." ), ) parser.add_argument( @@ -139,8 +140,8 @@ def handle(self, *args, **options): self.stdout.write( self.style.WARNING( "--skip-daily without --skip-monthly: the aggregation task " - "rebuilds the current and previous month from the daily tier " - "and will drop monthly rows this run writes there." + "rebuilds the current and previous month from the daily tier, " + "so monthly will be overwritten with an under-count." ) ) diff --git a/backend/dashboard_metrics/tasks.py b/backend/dashboard_metrics/tasks.py index 332cb623ef..f9b896f68a 100644 --- a/backend/dashboard_metrics/tasks.py +++ b/backend/dashboard_metrics/tasks.py @@ -14,7 +14,6 @@ from account_v2.models import Organization from celery import shared_task from django.core.cache import cache -from django.db import transaction from django.db.models import Min, Sum from django.db.models.functions import TruncMonth from django.db.utils import DatabaseError, OperationalError @@ -49,8 +48,11 @@ # in the schedule row's kwargs — editing this constant does not move the schedule. DASHBOARD_RECONCILE_WINDOW_DAYS = 7 -# Wider than the source window: metrics keyed on another column -# (e.g. approved_at) can land for an org whose executions are older. +# Floor on the prefilter lookback: metrics keyed on another column (e.g. +# approved_at) can land for an org whose executions are older. _active_org_ids +# takes the wider of this and the run's own window, so the prefilter is never +# narrower than what is being queried — at the 7-day reconciliation window the +# two are equal rather than this one being wider. DASHBOARD_ACTIVE_ORG_LOOKBACK_DAYS = 7 @@ -185,54 +187,17 @@ def _bulk_upsert_daily(aggregations: dict) -> int: return len(objects) -def _delete_orphan_monthly(objects: list[EventMetricsMonthly]) -> int: - """Drop stale monthly rows inside the partitions the rollup actually covered. - - Monthly derives from the daily tier, so a key with no daily rows left must - not survive as a stale total. An (organization, month) the rollup produced - nothing for is left alone instead: absent daily rows there mean an - incomplete tier, not a metric that went to zero. - """ - fresh_keys = { - (o.organization_id, o.month, o.metric_name, o.project, o.tag) for o in objects - } - covered: dict[date, set] = {} - for o in objects: - covered.setdefault(o.month, set()).add(o.organization_id) - - deleted = 0 - for month, org_ids in covered.items(): - stale_pks = [ - row["pk"] - for row in EventMetricsMonthly._base_manager.filter( - month=month, organization_id__in=org_ids - ).values("pk", "organization_id", "month", "metric_name", "project", "tag") - if ( - row["organization_id"], - row["month"], - row["metric_name"], - row["project"], - row["tag"], - ) - not in fresh_keys - ] - if not stale_pks: - continue - removed, _ = EventMetricsMonthly._base_manager.filter(pk__in=stale_pks).delete() - deleted += removed - - return deleted - - -def _rollup_monthly_from_daily(month_start: date) -> tuple[int, int]: +def _rollup_monthly_from_daily(month_start: date) -> int: """Sum the daily tier from month_start into monthly, for all orgs at once. + Upsert-only, per the design agreed on UN-3973: a monthly row the daily tier + no longer produces is left in place rather than deleted. A stale total is + recoverable with backfill_metrics; a deleted one is not, because the daily + rows that would rebuild it are exactly what is missing. + metric_type is aggregated rather than grouped: it is not part of unique_monthly_metric, so grouping on it could yield two rows for one conflict target. - - Returns: - (rows upserted, rows deleted as orphans) """ rows = ( EventMetricsDaily._base_manager.filter(date__gte=month_start) @@ -259,30 +224,36 @@ def _rollup_monthly_from_daily(month_start: date) -> tuple[int, int]: for row in rows ] - # Nothing to write and nothing covered, so nothing to sweep. The scoping in - # _delete_orphan_monthly already makes this safe; returning early just skips a - # pointless transaction. if not objects: - return 0, 0 - - with transaction.atomic(): - EventMetricsMonthly._base_manager.bulk_create( - objects, - update_conflicts=True, - unique_fields=["organization", "month", "metric_name", "project", "tag"], - update_fields=["metric_type", "metric_value", "metric_count"], - batch_size=MONTHLY_ROLLUP_BATCH_SIZE, - ) - deleted = _delete_orphan_monthly(objects) + return 0 - return len(objects), deleted + EventMetricsMonthly._base_manager.bulk_create( + objects, + update_conflicts=True, + unique_fields=["organization", "month", "metric_name", "project", "tag"], + update_fields=["metric_type", "metric_value", "metric_count"], + batch_size=MONTHLY_ROLLUP_BATCH_SIZE, + ) + return len(objects) -AGGREGATION_LOCK_KEY = "dashboard_metrics:aggregation_lock" +AGGREGATION_LOCK_KEY_PREFIX = "dashboard_metrics:aggregation_lock" AGGREGATION_LOCK_TIMEOUT = 900 # 15 minutes (matches task schedule) -def _acquire_aggregation_lock() -> bool: +def _aggregation_lock_key(source_window_days: int) -> str: + """One key per schedule, not one key for the task. + + 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. + """ + return f"{AGGREGATION_LOCK_KEY_PREFIX}:{source_window_days}d" + + +def _acquire_aggregation_lock(lock_key: str) -> bool: """Acquire the distributed aggregation lock with self-healing. Stores a Unix timestamp as the lock value. If a previous run crashed @@ -295,22 +266,22 @@ def _acquire_aggregation_lock() -> bool: now = time.time() # Fast path: lock is free - if cache.add(AGGREGATION_LOCK_KEY, str(now), AGGREGATION_LOCK_TIMEOUT): + if cache.add(lock_key, str(now), AGGREGATION_LOCK_TIMEOUT): return True # Lock exists — check if it's stale (previous run died without releasing) - lock_value = cache.get(AGGREGATION_LOCK_KEY) + lock_value = cache.get(lock_key) if lock_value is None: # Expired between our check and get — lock is now free, try to acquire it - return cache.add(AGGREGATION_LOCK_KEY, str(now), AGGREGATION_LOCK_TIMEOUT) + return cache.add(lock_key, str(now), AGGREGATION_LOCK_TIMEOUT) try: lock_time = float(lock_value) except (TypeError, ValueError): # Corrupted value (e.g. old "running" string) — reclaim it logger.warning("Reclaiming aggregation lock with invalid value: %s", lock_value) - cache.delete(AGGREGATION_LOCK_KEY) - return cache.add(AGGREGATION_LOCK_KEY, str(now), AGGREGATION_LOCK_TIMEOUT) + cache.delete(lock_key) + return cache.add(lock_key, str(now), AGGREGATION_LOCK_TIMEOUT) age = now - lock_time if age > AGGREGATION_LOCK_TIMEOUT: @@ -319,8 +290,8 @@ def _acquire_aggregation_lock() -> bool: age, AGGREGATION_LOCK_TIMEOUT, ) - cache.delete(AGGREGATION_LOCK_KEY) - return cache.add(AGGREGATION_LOCK_KEY, str(now), AGGREGATION_LOCK_TIMEOUT) + cache.delete(lock_key) + return cache.add(lock_key, str(now), AGGREGATION_LOCK_TIMEOUT) return False @@ -350,14 +321,26 @@ def aggregate_metrics_from_sources( Returns: Dict with aggregation summary for all three tiers """ - if not _acquire_aggregation_lock(): - logger.info("Skipping aggregation — another run is in progress") - return {"success": True, "skipped": True, "reason": "lock_held"} + source_window_days = _validate_source_window(source_window_days) + lock_key = _aggregation_lock_key(source_window_days) + + if not _acquire_aggregation_lock(lock_key): + logger.warning( + "Skipping the %d-day aggregation — another run of the same schedule is " + "in progress", + source_window_days, + ) + return { + "success": True, + "skipped": True, + "reason": "lock_held", + "source_window_days": source_window_days, + } try: return _run_aggregation(source_window_days) finally: - cache.delete(AGGREGATION_LOCK_KEY) + cache.delete(lock_key) def _aggregate_single_metric( @@ -572,7 +555,10 @@ def _build_result( ) -> dict[str, Any]: """Shape the task's return value from the accumulated stats.""" result = { - "success": True, + # 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. + "success": stats["errors"] == 0, "organizations_processed": stats["orgs_processed"], "hourly": stats["hourly"], "daily": stats["daily"], @@ -589,10 +575,33 @@ def _build_result( return result +# A negative window puts daily_start in the future so nothing matches; 0 never +# refreshes yesterday; an unbounded one restores the multi-month per-org scan this +# change exists to remove, past soft_time_limit. +MAX_SOURCE_WINDOW_DAYS = 90 + + +def _validate_source_window(source_window_days: int) -> int: + """Coerce and bound the window. It arrives as JSON from an editable Beat row.""" + try: + days = int(source_window_days) + except (TypeError, ValueError) as exc: + raise ValueError( + f"source_window_days must be an integer, got {source_window_days!r}" + ) from exc + if not 1 <= days <= MAX_SOURCE_WINDOW_DAYS: + raise ValueError( + f"source_window_days must be between 1 and {MAX_SOURCE_WINDOW_DAYS}, " + f"got {days}" + ) + return days + + def _run_aggregation( source_window_days: int = DASHBOARD_SOURCE_WINDOW_DAYS, ) -> dict[str, Any]: """Execute the aggregation, separately from the task's lock handling.""" + source_window_days = _validate_source_window(source_window_days) end_date = timezone.now() # Monthly spans the current and previous month. @@ -605,7 +614,7 @@ def _run_aggregation( stats = { "hourly": {"upserted": 0}, "daily": {"upserted": 0}, - "monthly": {"upserted": 0, "deleted": 0}, + "monthly": {"upserted": 0, "failed": False}, "errors": 0, "orgs_processed": 0, } @@ -640,27 +649,24 @@ def _run_aggregation( stats["errors"] += 1 try: - upserted, deleted = _rollup_monthly_from_daily(monthly_start) - stats["monthly"]["upserted"] = upserted - stats["monthly"]["deleted"] = deleted - if deleted: - logger.warning( - "Monthly rollup deleted %d orphan row(s) from %s", deleted, monthly_start - ) + 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 - logger.info( + log = logger.warning if stats["errors"] else logger.info + log( f"Aggregation completed: {stats['orgs_processed']} orgs, " f"hourly={stats['hourly']['upserted']}, " f"daily={stats['daily']['upserted']}, " f"monthly={stats['monthly']['upserted']}, " - f"monthly_deleted={stats['monthly']['deleted']}, " f"errors={stats['errors']}" ) diff --git a/backend/dashboard_metrics/tests/test_tasks.py b/backend/dashboard_metrics/tests/test_tasks.py index e85acd203b..e7911226d6 100644 --- a/backend/dashboard_metrics/tests/test_tasks.py +++ b/backend/dashboard_metrics/tests/test_tasks.py @@ -1,12 +1,14 @@ """Unit tests for Dashboard Metrics Celery tasks.""" import json +import time from datetime import date, datetime, timedelta from importlib import import_module from types import SimpleNamespace from unittest.mock import patch from django.apps import apps +from django.core.cache import cache from django.db import connection from django.db.utils import DatabaseError from django.test import TestCase @@ -19,6 +21,7 @@ EventMetricsDaily, EventMetricsHourly, EventMetricsMonthly, + Granularity, MetricType, ) from pg_queue.models import PgPeriodicTask @@ -27,15 +30,20 @@ from workflow_manager.workflow_v2.models.execution import WorkflowExecution from workflow_manager.workflow_v2.models.workflow import Workflow from dashboard_metrics.internal_views import AggregateMetricsAPIView +from dashboard_metrics.services import MetricsQueryService from dashboard_metrics.tasks import ( + AGGREGATION_LOCK_TIMEOUT, DASHBOARD_RECONCILE_WINDOW_DAYS, DASHBOARD_SOURCE_WINDOW_DAYS, + _acquire_aggregation_lock, _active_org_ids, + _aggregation_lock_key, _rollup_monthly_from_daily, _run_aggregation, _truncate_to_day, _truncate_to_hour, _truncate_to_month, + _validate_source_window, aggregate_metrics_from_sources, cleanup_daily_metrics, cleanup_hourly_metrics, @@ -264,7 +272,7 @@ def test_sums_daily_rows_into_month_bucket(self): self._daily(date(2024, 3, 5), value=10, count=2) self._daily(date(2024, 3, 18), value=32, count=4) - assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (1, 0) + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == 1 rows = self._monthly_rows() assert len(rows) == 1 @@ -279,7 +287,7 @@ def test_month_boundary_keeps_months_separate(self): self._daily(date(2024, 2, 1), value=100) self._daily(date(2024, 2, 2), value=200) - assert _rollup_monthly_from_daily(date(2024, 1, 1)) == (2, 0) + assert _rollup_monthly_from_daily(date(2024, 1, 1)) == 2 rows = self._monthly_rows() assert [r.month for r in rows] == [date(2024, 1, 1), date(2024, 2, 1)] @@ -290,7 +298,7 @@ def test_excludes_months_before_the_window(self): self._daily(date(2023, 12, 15), value=999) self._daily(date(2024, 1, 15), value=5) - assert _rollup_monthly_from_daily(date(2024, 1, 1)) == (1, 0) + assert _rollup_monthly_from_daily(date(2024, 1, 1)) == 1 rows = self._monthly_rows() assert len(rows) == 1 @@ -314,7 +322,7 @@ def test_mixed_metric_type_within_a_month_yields_one_row(self): self._daily(date(2024, 3, 5), value=10, metric_type=MetricType.HISTOGRAM) self._daily(date(2024, 3, 6), value=5, metric_type=MetricType.COUNTER) - assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (1, 0) + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == 1 rows = self._monthly_rows() assert len(rows) == 1 @@ -324,8 +332,8 @@ def test_an_empty_daily_tier_leaves_existing_monthly_rows_alone(self): """An empty tier means the source is gone, not that every month is zero. Seeding a monthly row first is what makes the failure reachable at all: with - an empty table a sweep that deletes everything and one that deletes nothing - both leave an empty table, and the assertion passes either way. + an empty table an implementation that wipes and one that writes nothing both + leave an empty table, and the assertion passes either way. """ EventMetricsMonthly._base_manager.create( organization=self.org, @@ -337,53 +345,58 @@ def test_an_empty_daily_tier_leaves_existing_monthly_rows_alone(self): project="default", ) - assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (0, 0) + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == 0 rows = self._monthly_rows() assert len(rows) == 1 assert rows[0].metric_value == 42 - def test_metric_is_dropped_once_its_daily_rows_are_gone(self): - """A metric with no daily rows left must not keep a stale monthly total.""" + def test_a_metric_whose_daily_rows_are_gone_keeps_its_last_total(self): + """Upsert-only, per the design agreed on UN-3973. + + A stale total is recoverable — backfill_metrics rewrites it. A deleted row is + not, because the daily rows that would rebuild it are exactly what is missing. + """ self._daily(date(2024, 3, 5), value=10) self._daily(date(2024, 3, 6), value=7, metric_name="pages_processed") - assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (2, 0) + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == 2 EventMetricsDaily._base_manager.filter(metric_name="pages_processed").delete() - assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (1, 1) + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == 1 rows = self._monthly_rows() - assert [r.metric_name for r in rows] == ["documents_processed"] + assert [r.metric_name for r in rows] == ["documents_processed", "pages_processed"] - def test_a_month_with_no_daily_rows_of_its_own_is_left_alone(self): - """An incomplete daily tier must read as "unknown", not as "deleted". + def test_a_partially_repopulated_month_is_overwritten_not_accumulated(self): + """The realistic post-downtime shape: the daily tier comes back short. - A partially populated tier passes the empty-tier guard, so without scoping the - sweep to the (organization, month) partitions the rollup actually produced, - one missing month wipes that month's monthly rows for every org. + The total tracks whatever the daily tier currently holds, so repairing daily + repairs monthly on the next run — which is what makes upsert-only recoverable. """ self._daily(date(2024, 3, 5), value=10) - self._daily(date(2024, 4, 5), value=7) - assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (2, 0) + self._daily(date(2024, 3, 6), value=32) + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == 1 + assert self._monthly_rows()[0].metric_value == 42 - EventMetricsDaily._base_manager.filter(date=date(2024, 3, 5)).delete() - assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (1, 0) + EventMetricsDaily._base_manager.filter(date=date(2024, 3, 6)).delete() + _rollup_monthly_from_daily(date(2024, 3, 1)) + assert self._monthly_rows()[0].metric_value == 10 - months = [row.month for row in self._monthly_rows()] - assert months == [date(2024, 3, 1), date(2024, 4, 1)] + self._daily(date(2024, 3, 6), value=32) + _rollup_monthly_from_daily(date(2024, 3, 1)) + assert self._monthly_rows()[0].metric_value == 42 - def test_the_sweep_is_scoped_per_organization(self): - """The sweep bypasses the org-scoped manager, so the org half of the key is - load-bearing: drop it and one org's daily rows vouch for another's monthly.""" + def test_rows_for_other_organizations_are_never_touched(self): + """The rollup goes through _base_manager, bypassing the org-scoped default.""" other = Organization.objects.create( organization_id="rollup-org-2", name="rollup-org-2", display_name="Other" ) self._daily(date(2024, 3, 5), value=10) self._daily(date(2024, 3, 6), value=20, org=other) - assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (2, 0) + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == 2 EventMetricsDaily._base_manager.filter(organization=other).delete() - assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (1, 0) + assert _rollup_monthly_from_daily(date(2024, 3, 1)) == 1 rows = self._monthly_rows() assert [(r.organization_id, r.metric_value) for r in rows] == [ @@ -391,28 +404,6 @@ def test_the_sweep_is_scoped_per_organization(self): (other.id, 20), ] - def test_a_stale_row_is_dropped_for_one_organization_only(self): - """Two orgs in one month: only the org whose metric vanished loses its row.""" - other = Organization.objects.create( - organization_id="rollup-org-3", name="rollup-org-3", display_name="Other" - ) - self._daily(date(2024, 3, 5), value=10) - self._daily(date(2024, 3, 5), value=1, metric_name="pages_processed") - self._daily(date(2024, 3, 6), value=20, org=other) - self._daily(date(2024, 3, 6), value=2, metric_name="pages_processed", org=other) - assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (4, 0) - - EventMetricsDaily._base_manager.filter( - organization=other, metric_name="pages_processed" - ).delete() - assert _rollup_monthly_from_daily(date(2024, 3, 1)) == (3, 1) - - survivors = { - (r.organization_id, r.metric_name) for r in self._monthly_rows() - } - assert (self.org.id, "pages_processed") in survivors - assert (other.id, "pages_processed") not in survivors - def test_months_before_the_window_are_left_alone(self): """Orphan cleanup must not reach outside the rebuilt window.""" self._daily(date(2024, 1, 10), value=99) @@ -506,8 +497,9 @@ def test_a_database_error_propagates_instead_of_reporting_success(self): with self.assertRaises(DatabaseError): _run_aggregation() - def test_an_unexpected_error_is_still_counted_rather_than_fatal(self): - """Everything outside the retry set keeps the previous non-fatal posture.""" + def test_an_unexpected_error_is_counted_but_does_not_abort_the_run(self): + """Everything outside the retry set stays non-fatal — the hourly and daily + tiers this run already wrote are kept — but it is not reported as success.""" with patch("dashboard_metrics.tasks.WorkflowExecution") as mock_execution: prefilter = mock_execution.objects.filter.return_value prefilter.values_list.return_value.distinct.return_value = [1] @@ -516,8 +508,9 @@ def test_an_unexpected_error_is_still_counted_rather_than_fatal(self): side_effect=ValueError("bad row"), ): result = _run_aggregation() - assert result["success"] is True + assert result["success"] is False assert result["errors"] == 1 + assert result["monthly"]["failed"] is True class TestInternalAggregateEndpoint(TestCase): @@ -613,7 +606,7 @@ def test_the_window_covers_the_previous_month_and_spares_what_precedes_it(self): result = self._run() assert result["period"]["monthly"]["start"] == self.last_month.isoformat() - assert result["monthly"] == {"upserted": 2, "deleted": 0} + assert result["monthly"] == {"upserted": 2, "failed": False} rows = EventMetricsMonthly._base_manager.order_by("month") assert [r.month for r in rows] == [ @@ -622,14 +615,212 @@ def test_the_window_covers_the_previous_month_and_spares_what_precedes_it(self): self.this_month, ] - def test_the_run_reports_what_it_deleted(self): - """The only destructive write in the task has to reach the result.""" + def test_a_failed_rollup_is_not_reported_as_nothing_to_do(self): + """upserted stays 0 on failure, which is also the legitimate empty value. + + Three states used to collapse into one alongside success: True — failed, + empty, and no active orgs. + """ self._daily(self.this_month, value=10) - self._daily(self.this_month, value=1, metric_name="pages_processed") - assert self._run()["monthly"] == {"upserted": 2, "deleted": 0} + with patch( + "dashboard_metrics.tasks._rollup_monthly_from_daily", + side_effect=ValueError("bad row"), + ): + result = self._run() - EventMetricsDaily._base_manager.filter(metric_name="pages_processed").delete() - assert self._run()["monthly"] == {"upserted": 1, "deleted": 1} + assert result["monthly"] == {"upserted": 0, "failed": True} + assert result["success"] is False + + +class TestMonthlyMatchesTheOldDerivation(TestCase): + """AC-4: the new monthly figures equal the ones the source queries produced. + + Every other monthly test feeds hand-written daily rows in and checks the sum of + what it just wrote — self-consistency, not equivalence. This one seeds *source* + rows, lets the real aggregation populate the daily tier from them, and compares + the rolled-up monthly against the pre-change derivation computed independently: + `get_documents_processed` at DAY granularity, bucketed by month in Python. + + The window is deliberately wide enough to cover both months, which is the state + `backfill_metrics` establishes before this change is deployed. + """ + + def setUp(self): + self.org = Organization.objects.create( + organization_id="golden-org", name="golden-org", display_name="Golden Org" + ) + self.workflow = Workflow.objects.create( + workflow_name="golden-wf", organization=self.org + ) + # Offsets are derived from the month boundary, never fixed day counts: on the + # 25th of a month a hardcoded "25 days ago" lands in the current month and the + # cross-boundary coverage silently disappears. + now = timezone.now() + first_of_this_month = _truncate_to_month(now) + self.days_to_last_month_end = (now - first_of_this_month).days + 1 + self.days_to_last_month_start = ( + now - _truncate_to_month(first_of_this_month - timedelta(days=1)) + ).days + + def _seed(self, days_ago: int, count: int) -> None: + """Seed `count` completed file executions dated `days_ago`.""" + stamp = timezone.now() - timedelta(days=days_ago) + for n in range(count): + execution = WorkflowExecution.objects.create( + workflow=self.workflow, status=ExecutionStatus.COMPLETED + ) + file_execution = WorkflowFileExecution.objects.create( + workflow_execution=execution, + file_name=f"{days_ago}-{n}.pdf", + status=ExecutionStatus.COMPLETED.value, + ) + WorkflowFileExecution.objects.filter(pk=file_execution.pk).update( + created_at=stamp + ) + WorkflowExecution.objects.filter(pk=execution.pk).update(created_at=stamp) + + def _written(self) -> dict: + """Monthly totals as the rollup wrote them.""" + return { + row.month: row.metric_value + for row in EventMetricsMonthly._base_manager.filter( + metric_name="documents_processed" + ) + } + + def _oracle(self, monthly_start, end_date) -> dict: + """Monthly totals the way the code derived them before this change.""" + rows = MetricsQueryService.get_documents_processed( + organization_id=str(self.org.id), # tasks.py passes the numeric PK + start_date=monthly_start, + end_date=end_date, + granularity=Granularity.DAY, + ) + totals: dict = {} + for row in rows: + month = _truncate_to_month(row["period"]).date() + totals[month] = totals.get(month, 0) + row["value"] + return totals + + def test_monthly_equals_the_pre_change_figures_across_a_month_boundary(self): + self._seed(days_ago=0, count=3) + self._seed(days_ago=self.days_to_last_month_end, count=2) + self._seed(days_ago=self.days_to_last_month_start, count=4) + + with patch("dashboard_metrics.tasks.WorkflowExecution") as mock_execution: + prefilter = mock_execution.objects.filter.return_value + prefilter.values_list.return_value.distinct.return_value = [self.org.id] + result = _run_aggregation( + source_window_days=self.days_to_last_month_start + 1 + ) + + monthly_start = date.fromisoformat(result["period"]["monthly"]["start"]) + end_date = datetime.fromisoformat(result["period"]["monthly"]["end"]) + expected = self._oracle( + datetime.combine(monthly_start, datetime.min.time(), tzinfo=end_date.tzinfo), + end_date, + ) + + assert len(expected) == 2, f"fixture must straddle a month boundary: {expected}" + assert self._written() == expected + + def test_the_comparison_can_fail_when_the_daily_tier_is_wrong(self): + """Guards the test above: an oracle that always matches proves nothing. + + Monthly is the sum of whatever the daily tier holds, so corrupting a day has + to move the monthly total away from the source-derived figure. Corrupting + rather than deleting is the point — under upsert-only a *deleted* day leaves + the previous monthly total in place, which is the recoverable state the + design accepts and is covered by TestMonthlyRollup. + """ + self._seed(days_ago=0, count=3) + self._seed(days_ago=self.days_to_last_month_end, count=2) + + with patch("dashboard_metrics.tasks.WorkflowExecution") as mock_execution: + prefilter = mock_execution.objects.filter.return_value + prefilter.values_list.return_value.distinct.return_value = [self.org.id] + result = _run_aggregation( + source_window_days=self.days_to_last_month_start + 1 + ) + + monthly_start = date.fromisoformat(result["period"]["monthly"]["start"]) + end_date = datetime.fromisoformat(result["period"]["monthly"]["end"]) + expected = self._oracle( + datetime.combine(monthly_start, datetime.min.time(), tzinfo=end_date.tzinfo), + end_date, + ) + assert self._written() == expected + + last_month_day = ( + timezone.now() - timedelta(days=self.days_to_last_month_end) + ).date() + corrupted = EventMetricsDaily._base_manager.filter( + date=last_month_day, metric_name="documents_processed" + ).update(metric_value=99) + assert corrupted, "fixture wrote no daily row for the previous month" + + _rollup_monthly_from_daily(monthly_start) + assert self._written() != expected + + +class TestTheLockIsPerSchedule(TestCase): + """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) + + def test_a_held_key_does_not_block_the_other_schedule(self): + cache.clear() + assert _acquire_aggregation_lock( + _aggregation_lock_key(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) + ) + # The reconciliation pass proceeds regardless. + assert _acquire_aggregation_lock( + _aggregation_lock_key(DASHBOARD_RECONCILE_WINDOW_DAYS) + ) + cache.clear() + + def test_a_stale_lock_is_reclaimed(self): + cache.clear() + key = _aggregation_lock_key(DASHBOARD_SOURCE_WINDOW_DAYS) + 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) + cache.set(key, "running", 3600) + assert _acquire_aggregation_lock(key) + cache.clear() + + +class TestSourceWindowValidation(TestCase): + """The window arrives as JSON from a Beat row editable in the admin.""" + + def test_a_sane_window_passes_through(self): + assert _validate_source_window(7) == 7 + assert _validate_source_window("7") == 7 + + def test_a_window_that_would_query_nothing_is_rejected(self): + # Negative puts daily_start in the future; 0 never refreshes yesterday. + for bad in (-1, 0): + with self.assertRaises(ValueError): + _validate_source_window(bad) + + def test_a_window_that_restores_the_multi_month_scan_is_rejected(self): + with self.assertRaises(ValueError): + _validate_source_window(365) + + def test_a_non_integer_window_is_rejected(self): + with self.assertRaises(ValueError): + _validate_source_window("seven") class TestSourceWindow(TestCase): diff --git a/workers/scheduler/dashboard_metrics_tasks.py b/workers/scheduler/dashboard_metrics_tasks.py index 07787aa9f3..90e016682a 100644 --- a/workers/scheduler/dashboard_metrics_tasks.py +++ b/workers/scheduler/dashboard_metrics_tasks.py @@ -92,16 +92,27 @@ def _call_internal( def _log_if_skipped(name: str, result: dict[str, Any]) -> None: - """Surface a lock-held no-op. + """Surface a run that did nothing, whatever shape the backend reported it in. - The backend returns success with ``skipped=True`` when the Redis lock is held. That - is correct behaviour, but left at INFO a permanently leaked lock looks like 96 - successful runs a day that did nothing. + Three of them, and only the first sets ``skipped``: the Redis lock was held + (``skipped``/``reason``), no organisation had recent activity + (``skipped_reason``), or every metric raised and was caught per-metric + (``errors``). Each is correct behaviour in isolation, but left at INFO a leaked + lock or a frozen source table looks like 96 successful runs a day. """ if result.get("skipped"): logger.warning( "%s did no work: %s", name, result.get("reason", "reported skipped=True") ) + elif result.get("skipped_reason"): + logger.warning("%s did no work: %s", name, result["skipped_reason"]) + elif result.get("errors"): + logger.warning( + "%s completed with %s error(s) across %s organisation(s)", + name, + result["errors"], + result.get("organizations_processed", "?"), + ) @worker_task(name="dashboard_metrics.aggregate_from_sources")