Skip to content

UN-4046 [DOCS] Document the PG queue transport in the dashboard-metrics README #2280

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kirtimanmishrazipstack merged 3 commits into from
Sep 10, 2026
Merged

UN-4046 [DOCS] Document the PG queue transport in the dashboard-metrics README #2280

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1d3cb4c
docs: document the PG queue transport in the dashboard-metrics README
jaseemjaskp Sep 10, 2026
0fa100f
docs: name worker-pg-reaper as the firer, and reconcile the retry claim
jaseemjaskp Sep 10, 2026
a32e7a1
docs: correct the drift claim, the --migrate gate, and the rollback r…
jaseemjaskp Sep 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
231 changes: 198 additions & 33 deletions backend/dashboard_metrics/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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/)
Expand All @@ -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) |
Expand All @@ -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`) |
Expand Down Expand Up @@ -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)

Expand Down Expand Up @@ -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.

---
Expand Down Expand Up @@ -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`) |
Expand All @@ -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 = [
Expand All @@ -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:
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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

Expand All @@ -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
Expand All @@ -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}')
"
Expand Down Expand Up @@ -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"},
```
Expand Down Expand Up @@ -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
Expand All @@ -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
```

Expand Down
Loading