diff --git a/backend/dashboard_metrics/README.md b/backend/dashboard_metrics/README.md index a80e5ded63..61a61c40dc 100644 --- a/backend/dashboard_metrics/README.md +++ b/backend/dashboard_metrics/README.md @@ -14,7 +14,8 @@ This module provides a metrics dashboard for monitoring document processing, API ### Data Flow ``` Source Tables (usage_v2, page_usage, workflow_execution, workflow_file_execution) - ↓ [Celery: hourly tier every 15 min, daily+monthly hourly at :20] + ↓ [PG scheduler fires; hourly tier every 15 min, daily+monthly hourly at :20] + ↓ [worker-pg-metrics drains dashboard_metric_events → POSTs the internal API] Aggregated Tables (EventMetricsHourly → Daily → Monthly) ↓ API Endpoints (/overview/, /summary/, /series/) @@ -26,7 +27,8 @@ Frontend Dashboard (MetricsSummary, MetricsChart, MetricsTable) | File | Purpose | |------|---------| | `services.py` | Queries source tables (9 methods, one per metric) | -| `tasks.py` | Celery tasks for aggregation and cleanup | +| `tasks.py` | The aggregation and cleanup task bodies (run by both transports) | +| `internal_views.py` | Internal API the PG consumer POSTs to, which calls `tasks.py` | | `views.py` | REST API endpoints | | `cache.py` | Redis caching layer (bucket-based MGET) | | `models.py` | Aggregated tables (Hourly/Daily/Monthly) | @@ -39,14 +41,17 @@ Frontend Dashboard (MetricsSummary, MetricsChart, MetricsTable) # --skip-hourly still issues the HOUR queries it skips here. python manage.py backfill_metrics --days=30 -# Start metrics worker -celery -A backend worker -Q dashboard_metric_events -l info +# PG transport (the default; see "Transport" below) — BOTH are required. +# The consumer drains the queue; the reaper is what actually fires the schedule. +(cd workers/ && ./run-worker.sh pg-metrics) +(cd workers/ && ./run-worker.sh reaper) -# Start scheduler (for periodic aggregation) +# Legacy Celery transport — rollback only, needs PG_SCHEDULER_ENABLED=false +celery -A backend worker -Q dashboard_metric_events -l info celery -A backend beat -l info ``` -### Celery Tasks & Schedule +### Scheduled Tasks & Schedule | Task | Schedule | What It Does | |------|----------|--------------| | `aggregate_from_sources` | Every 15 min | Aggregates source → **hourly tier only** (`tier=hourly`) | @@ -77,10 +82,11 @@ celery -A backend beat -l info - [Metrics Definitions](#metrics-definitions) - [Source Tables](#source-tables) - [Aggregated Tables](#aggregated-tables) -- [Celery Tasks](#celery-tasks) +- [Scheduled Tasks](#scheduled-tasks) +- [Transport: how the cron actually fires](#transport-how-the-cron-actually-fires) - [API Endpoints](#api-endpoints) - [Caching Strategy](#caching-strategy) -- [Frontend Components](#frontend-components) +- [UI Data Flow](#ui-data-flow--what-shows-where) - [Setup & Configuration](#setup--configuration) - [Management Commands](#management-commands) @@ -170,7 +176,7 @@ The dashboard reads from **pre-aggregated tables** (`event_metrics_hourly`, `eve - If the aggregation task fails, the dashboard shows stale data rather than crashing — up to 15 minutes old for hourly figures, up to an hour for daily and monthly. - A daily 04:40 UTC reconciliation pass reruns the same task over a 7-day source window, so a **daily- or monthly-tier** gap shorter than that repairs itself without a manual backfill. The hourly tier re-queries the last 24h on every run regardless of tier or window, so an hourly gap shorter than 24h repairs itself on the next tick; only one older than 24h needs `backfill_metrics`. - The 7-day window is also the ceiling on lag, not just on downtime. `documents_processed` and `failed_pages` filter on a terminal status but window and bucket on `created_at`, so a row whose status turns terminal more than 7 days after it was created is counted in no daily row — and therefore in no monthly total either, since monthly is the sum of daily. Before the monthly tier was derived from daily this was caught by the wider monthly source window. -- Celery tasks have `max_retries=3` with exponential backoff. +- Retry behaviour depends on the transport: on Celery the tasks carry `max_retries=3` with exponential backoff; on the PG transport (the default) retries are governed by the consumer's `MAX_ATTEMPTS`, which compose sets to `1` — a failed run is dropped and the next scheduled tick supersedes it. See [Transport](#transport-how-the-cron-actually-fires). - Cleanup tasks (hourly: 30-day retention, daily: 365-day retention) prevent unbounded table growth. --- @@ -338,13 +344,18 @@ This enables upsert operations during aggregation. --- -## Celery Tasks +## Scheduled Tasks + +The task bodies live in `tasks.py` and carry `@shared_task`, but **which process fires +and runs them depends on the transport** — see +[Transport](#transport-how-the-cron-actually-fires) below. The task name, the queue and +the schedule are the same on both. ### Task Configuration Located in `tasks.py`: -| Task Name | Celery Name | Schedule | Queue | Purpose | +| Python Function | Registered Task Name | Schedule | Queue | Purpose | |-----------|-------------|----------|-------|---------| | `aggregate_metrics_from_sources` | `dashboard_metrics.aggregate_from_sources` | Every 15 min | `dashboard_metric_events` | Aggregate the hourly tier (`tier=hourly`) | | `aggregate_metrics_from_sources` | `dashboard_metrics.aggregate_from_sources` | Hourly at :20 UTC | `dashboard_metric_events` | Aggregate the daily and monthly tiers (`tier=daily_monthly`) | @@ -354,7 +365,9 @@ Located in `tasks.py`: ### Queue Configuration -In `backend/celery_config.py`: +The queue name `dashboard_metric_events` is shared by both transports: on the PG +transport it is a `pg_queue_message` queue, on the Celery transport a RabbitMQ one. +The routing below is the **Celery** half, in `backend/celery_config.py`: ```python task_queues = [ @@ -369,16 +382,6 @@ task_routes = { } ``` -### Running the Worker - -```bash -# Start the dashboard metrics worker -celery -A backend worker --loglevel=info -Q dashboard_metric_events - -# Start Celery Beat for periodic tasks -celery -A backend beat --loglevel=info -``` - ### Aggregation Task Details The `aggregate_metrics_from_sources` task: @@ -480,6 +483,123 @@ the rollup derive monthly from it. --- +## Transport: how the cron actually fires + +There are two transports. **The PG queue is the default and the one a current deploy +uses**; Celery Beat is the rollback target. Both run the same function in the backend +process — the difference is what schedules it and what carries it. + +### Schedule rows are dual-written + +Every `dashboard_metrics.*` schedule exists as a **pair** of rows: a +`django_celery_beat.PeriodicTask` and a `pg_queue.PgPeriodicTask`. A row's `pg_owned` +flag says which scheduler fires it. + +How the pair is kept in step differs by row, which matters when you edit one: + +| Rows | Declared in | Drift risk | +|------|-------------|------------| +| The two aggregation rows, and the reconciliation row | One spec in a single migration — `0005_add_reconciliation_task.py`, `0006_split_aggregation_schedule.py` (`split_schedules`) | None by construction | +| The three original rows (`aggregate_from_sources`, both cleanups) | **Two separate migrations** — `0002_setup_periodic_tasks.py` (Beat) and its PG twin `0004_pg_periodic_tasks.py` | Real — edit one and forget the other and they diverge | + +The second row is guarded only by a test, `tests/test_pg_periodic_task_declarations.py`, +which exists for exactly that failure mode. If you change one of those three schedules, +change both migrations. + +### PG transport (default) + +```text +pg_periodic_task row (pg_owned=true, cron_string) + ↓ leader-elected orchestrator tick evaluates the cron — this is worker-pg-reaper + workers/queue_backend/pg_queue/reaper.py -> pg_scheduler.dispatch_due_periodic_tasks + ↓ enqueue + advance next_run_at IN ONE TRANSACTION (a crash cannot double-fire) +pg_queue_message on `dashboard_metric_events` + ↓ worker-pg-metrics (a generic pg-queue-consumer pinned to that queue) +workers/scheduler/dashboard_metrics_tasks.py ← an HTTP proxy, NOT the real work + ↓ POST v1/dashboard-metrics/aggregate/ (internal API auth, no X-Organization-ID) + relative to INTERNAL_API_BASE_URL, which ends /internal — the full path + is /internal/v1/dashboard-metrics/aggregate/ +backend/dashboard_metrics/internal_views.py + ↓ calls tasks.py verbatim +aggregate_metrics_from_sources(...) +``` + +The workers image has **no Django**, so the ORM-heavy aggregation cannot run there — +hence the proxy hop. The windows, the Redis lock and the upsert logic are reused +unchanged, not reimplemented. + +`worker-pg-metrics` is a dedicated service rather than another queue on +`worker-pg-scheduler` because the consumer's health heartbeat freezes while a task runs, +so `HEALTH_STALE` doubles as an upper bound on one task's wall clock. The aggregation +runs for minutes and would trip the scheduler's liveness bound (~4 min) where the +metrics consumer's is far higher (~15 min). Both are `${VAR:-default}` overrides — read +the current numbers from `docker/docker-compose.yaml` (`worker-pg-metrics`), and note +the values that apply in k8s come from the chart in a separate repo. + +### Celery transport (legacy / rollback) + +Beat reads the `PeriodicTask` row and publishes over RabbitMQ to +`celery -A backend worker -Q dashboard_metric_events`, which runs the task body +in-process. No proxy hop. + +### Which one is live + +`converge_pg_scheduler` decides, and `backend/entrypoint.sh` runs it on every backend +start **that runs migrations** — the whole block is gated on `--migrate`, which the +default compose stack passes. Ownership is then a values change rather than a remembered +procedure. **Where migrations run as a separate job or init-container rather than in the +backend's own start, this does not fire and ownership will not follow the env var.** + +| Variable | Default | Effect | +|----------|---------|--------| +| `PG_SCHEDULER_ENABLED` | **`true`** | PG scheduler owns the pipeline schedules; matching Beat rows are disabled | +| `PG_SCHEDULER_ADOPT_PERIODICS` | unset (`false`) | Must be `true` to also move the periodics, which is what carries the `dashboard_metrics.*` rows — it passes `--periodics`, and that flag covers **every** mirrored non-pipeline periodic, not only these | + +Both gates are required for the metrics cron. The periodics are separate because +adopting them needs `worker-pg-metrics` deployed first. The default compose stack sets +`PG_SCHEDULER_ADOPT_PERIODICS=true` and defines **no** `celery-beat` or `worker-metrics` +service at all. + +> **Rolling back is not just the env var — and do not unset both gates.** +> +> `PG_SCHEDULER_ADOPT_PERIODICS` must stay **`true`** while `PG_SCHEDULER_ENABLED` goes +> to `false`. It gates `--periodics`, which the release direction needs just as much as +> the adopt direction: without it `converge_pg_scheduler` never touches the +> `dashboard_metrics.*` rows at all. Turn both off together — the intuitive "roll it all +> back" — and those rows keep `pg_owned=True` with their Beat twins still disabled, +> which is the no-firer state below, reached by following the runbook. +> +> Converging to Beat also restores *ownership*, not *capacity*. Beat publishes to +> RabbitMQ, so a released schedule fires again only if `workerMetrics` is actually +> running. Flip it back on in the **same** change, or you have moved the outage rather +> than fixed it. + +### Failure modes worth knowing + +- On the PG transport a periodic is **fire-and-forget** — nothing records a task status, + so log severity is the only signal that reaches an alert. The proxy therefore splits + the two cases: a non-200 or a transport failure **raises** (marking the message + failed), while a 200 that reports per-org `errors` is logged at ERROR rather than + raised, since the run did happen. See `dashboard_metrics_tasks.py`. +- **Retries are effectively off, but that is a deployment setting, not a property of the + transport.** Compose sets `WORKER_PG_QUEUE_CONSUMER_MAX_ATTEMPTS=1`, so a failed + message is dropped and the next cron tick supersedes it; the consumer's own default is + 5. Check the value your environment sets before assuming either. Independently, the + proxy's `httpx` client retries connection *establishment* three times — it never + re-sends a request the server already received. +- The backend's `autoretry_for=(DatabaseError, OperationalError)` **does not retry on + the PG path.** The decorator is still in the call path — `internal_views.py` invokes + the decorated task object — but Celery's `retry()` re-raises instead of retrying when + a task is called directly rather than dispatched by a worker. Worth knowing before + reasoning about `throws=` or `on_failure` here. +- A schedule adopted with no **`worker-pg-reaper`** running, or released with + `workerMetrics` scaled to zero, has **no firer**: no error, no queue depth, no failed + pod. Check `pg_owned` against what is actually deployed. Note the service to check is + `worker-pg-reaper`, **not** `worker-pg-scheduler` — despite the name, the latter is + the pipeline-task consumer and fires no periodics. + +--- + ## API Endpoints ### Endpoint Overview @@ -732,7 +852,7 @@ uv sync ``` Required packages in `pyproject.toml`: -- `django-celery-beat` - For periodic task scheduling +- `django-celery-beat` - Holds the Beat half of every schedule row (see [Transport](#transport-how-the-cron-actually-fires)); still required even when the PG scheduler is the firer - `django-redis` - For MGET/pipeline operations ### 2. Run Migrations @@ -743,7 +863,7 @@ python manage.py migrate dashboard_metrics This creates: - Three aggregated tables -- Periodic task schedules in `django_celery_beat` +- Periodic task schedules — a paired row in **both** `django_celery_beat.PeriodicTask` and `pg_queue.PgPeriodicTask` (see [Transport](#transport-how-the-cron-actually-fires) for which migration declares which) ### 3. Backfill Historical Data @@ -760,6 +880,30 @@ python manage.py backfill_metrics --days=7 --dry-run ### 4. Start Workers +The PG transport is the default. It needs the metrics consumer, and the schedule rows +adopted onto the PG scheduler (`PG_SCHEDULER_ENABLED=true` **and** +`PG_SCHEDULER_ADOPT_PERIODICS=true`, which `entrypoint.sh` converges on a `--migrate` +start): + +```bash +# 1. The firer — the leader-elected loop that fires due pg_owned schedules. +# This is the Beat replacement; without it NOTHING is ever enqueued. +(cd workers/ && ./run-worker.sh reaper) + +# 2. The consumer — drains dashboard_metric_events off pg_queue. +(cd workers/ && ./run-worker.sh pg-metrics) +``` + +**Both are required**, the same way the Celery path needed Beat *and* a worker. The +scheduler tick is not a daemon of its own: it is leader-elected inside the +reaper/orchestrator loop (`reaper.py` calls `dispatch_due_periodic_tasks` each cycle). +In compose that loop is the `worker-pg-reaper` service, which is already declared — so +this second process is easy to miss when moving from compose to a local run, and +missing it produces exactly the silent no-firer state described under +[Transport](#transport-how-the-cron-actually-fires). + +Celery transport, for rollback only — requires `PG_SCHEDULER_ENABLED=false`: + ```bash # Terminal 1: Dashboard metrics worker celery -A backend worker --loglevel=info -Q dashboard_metric_events @@ -771,9 +915,16 @@ celery -A backend beat --loglevel=info ### 5. Verify Setup ```bash -# Check periodic tasks are registered +# Check periodic tasks are registered, and WHICH scheduler owns them. +# pg_owned=True -> the PG scheduler fires it (the Beat twin should be disabled) +# pg_owned=False -> Celery Beat fires it, and workerMetrics must be running python manage.py shell -c " from django_celery_beat.models import PeriodicTask +from pg_queue.models import PgPeriodicTask +print('-- pg_queue (default transport) --') +for t in PgPeriodicTask.objects.filter(name__startswith='dashboard'): + print(f'{t.name}: enabled={t.enabled} pg_owned={t.pg_owned} cron={t.cron_string}') +print('-- celery beat (rollback transport) --') for t in PeriodicTask.objects.filter(name__startswith='dashboard'): print(f'{t.name}: enabled={t.enabled}') " @@ -822,11 +973,19 @@ caveats that do not fit one line): **Fix**: 1. Check source tables have data 2. Run backfill command -3. Verify Celery worker is running - -#### 3. Celery task not running -**Cause**: Task not routed to correct queue -**Fix**: Ensure `celery_config.py` has the task route: +3. Verify the metrics consumer is running — `worker-pg-metrics` on the default + transport, `workerMetrics` if you have rolled back to Celery + +#### 3. Aggregation task never runs +**Cause (PG transport, most common)**: the schedule has **no firer**. This is silent — +no error, no queue depth, no failed pod. +**Fix**: check ownership matches what is deployed (see step 5 of +[Setup](#5-verify-setup)). `pg_owned=False` with no `workerMetrics` running, or +`pg_owned=True` with no PG scheduler, both fire nothing. Ownership is converged from +`PG_SCHEDULER_ENABLED` + `PG_SCHEDULER_ADOPT_PERIODICS` on backend start. + +**Cause (Celery transport)**: task not routed to the correct queue. +**Fix**: ensure `celery_config.py` has the task route: ```python "dashboard_metrics.aggregate_from_sources": {"queue": "dashboard_metric_events"}, ``` @@ -880,7 +1039,9 @@ backend/dashboard_metrics/ ├── models.py # EventMetricsHourly/Daily/Monthly models ├── serializers.py # DRF serializers ├── services.py # MetricsQueryService (source table queries) -├── tasks.py # Celery tasks (aggregation, cleanup) +├── tasks.py # Task bodies (aggregation, cleanup) — both transports +├── internal_urls.py # Internal API routes (PG transport) +├── internal_views.py # Internal API views the PG consumer POSTs to ├── urls.py # URL routing ├── views.py # API ViewSet ├── README.md # This file @@ -889,8 +1050,12 @@ backend/dashboard_metrics/ │ └── backfill_metrics.py # Backfill historical data ├── migrations/ │ ├── 0001_initial.py # Create 3 aggregation tables -│ └── 0002_setup_periodic_tasks.py # Register Celery Beat schedules -└── tests/ +│ ├── 0002_setup_periodic_tasks.py # Register the Beat schedules +│ ├── 0003_alter_eventmetricsdaily_organization_and_more.py +│ ├── 0004_pg_periodic_tasks.py # Mirror the schedules onto pg_queue +│ ├── 0005_add_reconciliation_task.py # Daily 04:40 UTC 7-day repair pass +│ └── 0006_split_aggregation_schedule.py # Split aggregation into two rows by tier +└── tests/ # Test suites (not exhaustively listed) └── test_tasks.py ```