From fe467d837e4be72069756e9f72b7ddbf3aa0159e Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Tue, 25 Aug 2026 12:00:49 +0530 Subject: [PATCH 1/2] 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/2] 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."""