[muhammad-ali-e]

Requested by ali · project thread

What

• New platform_metrics backend app serving GET /metrics in Prometheus text format (text/plain; version=0.0.4).
• Metrics, all unstract_-prefixed, no per-org labels:
  • unstract_queue_messages{queue,state} with state = ready | claimed | scheduled, and unstract_queue_oldest_message_age_seconds{queue} (from pg_queue_message)
  • unstract_workflow_executions{status} for pending / executing, and unstract_workflow_executions_recent{status} for completed / error / stopped that finished within a window (default 1h)
  • unstract_batches_in_progress, unstract_batches_stuck (from pg_barrier_state)
  • unstract_metrics_last_successful_refresh_timestamp{group}, unstract_metrics_refresh_failures_total{group} (one cluster-wide total kept in Redis), unstract_metrics_cache_up (0 while this process can't reach Redis)
• Known queues and statuses are always present and read 0 when empty. A queue outside the known list still shows up when it holds messages.

Why

• LLMWhisperer exposes queue metrics on its backend /metrics; Unstract had nothing equivalent. The PG reaper's /metrics doesn't split waiting from in-progress, has no execution counts, drops empty queues, and is empty on a standby reaper.

How

• Groups (groups.py): each MetricGroup has query() (DB, returns a JSON-safe snapshot) and families() (snapshot to Prometheus families). They're registered in METRIC_GROUPS, so a new group needs no change to the endpoint or the caching.
• Cache (cache.py): each group's snapshot lives in Redis for PLATFORM_METRICS_CACHE_TTL_SECONDS (15s), behind a non-blocking redis.lock. One pod refreshes a stale group and the rest serve the previous snapshot, so DB cost is about one refresh per group per 15s however many replicas get scraped. The cache is re-checked after the lock is taken, and the lock's expiry is sized from each group's statement count times the query timeout, so it can't expire mid-refresh.
• Resilience: a scrape always returns 200. If a refresh fails, the group serves its last snapshot and the failure counter goes up; the other groups aren't affected.
  • If Redis is down, or a Redis read or write fails, each pod serves its own newest copy and refreshes it at most once per TTL, so DB load stays bounded. The failure is still counted, and unstract_metrics_cache_up drops to 0.
  • Refresh failures are summed in Redis (HINCRBY + HGETALL in one transaction), so every pod and gunicorn worker reports the same monotonic total instead of a per-process count that would look like counter resets. While Redis is down the counter is left out of the scrape, and the buffered increments are flushed on the next successful one.
  • Each refresh runs under SET LOCAL statement_timeout (PLATFORM_METRICS_QUERY_TIMEOUT_MS, 5s), and the view is non_atomic_requests so that timeout never outlives the refresh.
• Consistency with the reaper: queue rows and oldest age use the same population as refresh_queue_gauges (all rows, now() - min(enqueued_at)), so the per-state counts add up to pg_queue_depth. batches_stuck is the reaper's stranded predicate, read from the same WORKER_PG_BATCH_STUCK_TIMEOUT_SECONDS. Queue metrics cover the PG transport only; work dispatched through Celery sits in RabbitMQ.
• Query cost (checked with EXPLAIN ANALYZE on 400k executions):
  • Active executions go through the partial we_active_by_workflow_idx.
  • Recent terminal executions are a we_created_at_idx range scan bounded by PLATFORM_METRICS_RECENT_LOOKBACK_SECONDS (24h). They are then filtered on the finish time, created_at + execution_time, which both status writers stamp on the terminal transition. modified_at is only the fallback, for rows terminalised without execution_time (the undispatched-execution sweep), because later writes such as result acknowledgement bump it.
  • The queue and barrier tables hold only in-flight rows, and the reaper already aggregates them the same way.
• Access:
  • The endpoint is mounted at the root next to /health. The public ingress only routes prefixed paths (/api/v1, /deployment, /public, …) to the backend, so /metrics is reachable in-cluster on the backend service only.
  • It is exempt from session auth (WHITELISTED_PATHS).
  • Setting PLATFORM_METRICS_TOKEN additionally requires Authorization: Bearer <token>, compared in constant time.

Can this PR break any existing features. If yes, please list possible items. If no, please explain why. (PS: Admins do not merge the PR without this section filled)

• No. It is a new route at a path no existing URL uses; the customer dashboard stays at /api/v1/unstract/<org>/metrics/. No execution path, model or migration changes. The only shared state is new Redis keys under platform_metrics:v1:*.

Database Migrations

• None.

Env Config

All optional:

• PLATFORM_METRICS_TOKEN (default empty, meaning no token check)
• PLATFORM_METRICS_CACHE_TTL_SECONDS (15)
• PLATFORM_METRICS_QUERY_TIMEOUT_MS (5000)
• PLATFORM_METRICS_RECENT_WINDOW_SECONDS (3600)
• PLATFORM_METRICS_RECENT_LOOKBACK_SECONDS (86400)
• PLATFORM_METRICS_QUEUES: comma-separated list of always-reported queues. It defaults to the queues the PG consumers in docker-compose.yaml consume. Cloud should override it if it runs extra PG queues.
• The existing WORKER_PG_BATCH_STUCK_TIMEOUT_SECONDS is now also read by the backend.

Relevant Docs

• Jira UN-4251. LLMWhisperer backend/app/controller/metrics.py was the reference pattern.

Related Issues or PRs

• UN-3672 (UN-3672 [FEAT] PG queue — application-level /metrics exporter (queue depth, message age, barriers, reaper outcomes) #2143), the reaper's /metrics, which this PR leaves unchanged. Endpoints for autoscaling are a separate ticket. PodMonitoring/scrape config in the cloud chart is out of scope here.

Dependencies Versions

• prometheus-client>=0.17.0,<1.0.0 added to the backend (same range the workers already pin; locks to 0.26.0).

Notes on Testing

• backend/platform_metrics/tests/, 43 tests:
  • 38 unit tests, with no DB or Redis. They cover:
    • output format and metric names
    • zero-filling
    • per-group failure isolation
    • cache/lock behaviour: fresh, stale, lock held elsewhere, re-check under the lock, Redis down, a failed re-read under the lock, a failed Redis write, an unreadable entry, lock sizing
    • the shared failure counter across processes, including a Redis outage and an unreadable hash value
    • access control
  • 5 DB tests (auto-marked integration) run against Postgres 16. They check each group's counts and edges:
    • the oldest age across states
    • window and lookback bounds, including a completion touched after it finished and swept rows
    • the stuck predicate, including remaining=0 lingerers
    • the statement timeout cancelling a slow query
• Smoke test with runserver against Postgres + Redis:
  • Requests without the token get 401.
  • Two scrapes in a row are served from the same Redis snapshot.
  • With Redis shut down, a scrape still returns 200 in about 10ms and the failure counter goes up.
• On a PodMonitoring setup, aggregate with max(), not sum(): every pod serves the same shared snapshot, and the failure counter is the same cluster-wide total on every pod.

Screenshots

Checklist

I have read and understood the Contribution Guidelines.

🤖 Generated with Claude Code

https://claude.ai/code/session_016Hov58G5waUqpMHz9P6FmL

[greptile-apps]

via Greptile

Confidence Score: 5/5

[Medium impact] The PR appears safe to merge; the previously reported execution-query failure is fixed.

Summary

Adds /metrics for queue, execution, and batch counts, with Redis-shared snapshots and optional bearer-token access.

• Backend scrapers can read platform metrics at /metrics.
• Scrapes show queue, workflow, and batch activity.

Diagram

%%{init: {'theme': 'neutral'}}%%
flowchart LR
  A["GET /metrics"] --> B["Optional bearer-token check"]
  B --> C["Select metric groups"]
  C --> D["Redis snapshot"]
  D --> E["Serve fresh or saved counts"]
  D --> F["Refresh stale counts under a lock"]
  F --> G["Postgres queries with a timeout"]
  G --> D
  D --> H["Use process-local copies if Redis fails"]
  E --> I["Prometheus text response"]
  H --> I

Loading

_{Reviews (5) · Last reviewed commit: "UN-4251 Type the execution-time duration..." · Reviewed by Greptile}

[CLAassistant]

All committers have signed the CLA.

[muhammad-ali-e]

Re Greptile P2 "Refresh failures counted twice" (cache.py:309-311): acknowledged, not changing this.

You're right that it can happen. Redis has to apply the MULTI/EXEC and then lose the reply on the way back, and in that case those increments are counted again on the next scrape. That window is narrow: it needs a connection drop between EXEC and its reply. The effect is also bounded and only goes up. The failure count over-reports by at most one scrape's worth of increments for this process.

The suggested fix is to store a cumulative count per process and add only the unseen delta. That needs a stable key for each process, and gunicorn recycles workers, so every recycle would leave behind a hash field that is never cleaned up. Summing those fields would also need the cleanup logic we'd have to build alongside it. For a diagnostic counter, where alerts should key on rate() > 0 or on unstract_metrics_cache_up == 0 rather than on exact totals, an occasional small over-count is the better trade than that bookkeeping.

---

Generated by Claude Code

[ritwik-g]

Thanks for this. I read it with its two likely readers in mind:

• GMP, once a PodMonitoring scrapes it, as LLMWhisperer's chart/llmwhisperer/templates/gcp/backend-monitoring.yaml does. This feeds the GCP dashboards and alerts.
• KEDA's metrics-api scaler, which we're about to point at unstract_queue_messages for the PG workers (UN-4525): one trigger per queue × ready/claimed, summed in a scalingModifiers formula.

I know UN-4251 asks for "serve its last cached values" on a failed refresh. Point 1 is about bounding that. Stale counts served as current mislead both readers.

1. Queue counts can be served indefinitely after refreshes start failing

When a refresh fails, the group serves its last snapshot (backend/platform_metrics/cache.py:228, exporter.py:60-62). The Redis copy is kept for an hour (cache.py:48, cache.py:180). Once that expires, _newest() returns the process-local copy (cache.py:259-263), which has no expiry (cache.py:112, cache.py:230). So if the backend can't run the aggregate (DB unreachable from the backend, or the 5 s statement_timeout), /metrics keeps answering 200 with the last good counts for the life of the pod.

Neither reader can tell a frozen count from a real one:

• GMP: the scrape succeeds, so up stays 1 and the frozen counts are stored as fresh points. Dashboards show a flat line, and a backlog alert keeps evaluating old numbers. The refresh timestamp is exported, but every panel and alert would have to be built against it to notice.
• KEDA: if the count froze low, workers don't scale up for new backlog, and the HPA can scale busy workers down to match it.

LLMWhisperer hit exactly this in LW-476. Zipstack/unstract-llm-whisperer#813 fixed it by omitting the per-queue series once the data is older than 60 s. KEDA then errors on the missing series (metric 'unstract_queue_messages' not found), the HPA holds its replicas, and the ScaledObject's fallback applies after failureThreshold errors.

A minimal version here:

• Add an optional max_staleness_seconds to MetricGroup. None keeps today's behaviour.
• Set it on QueueMetricGroup from PLATFORM_METRICS_QUEUE_MAX_STALENESS_SECONDS, default 60.
• In _GroupsCollector.collect(), skip group.families() when snapshot.age(now) is above it. Keep the unstract_metrics_last_successful_refresh_timestamp{group="queues"} sample, so the age stays visible, and keep the 200, so a Prometheus scrape and the other two groups are unaffected.

While the data is fresh, empty queues still read 0, so UN-4251's "no gaps" holds. A gap then means only that the count is unknown, which is what a dashboard should show and what KEDA's fallback is for.

60 s stays above the worst healthy age, which is the 15 s TTL plus the 30 s lock floor when a refresher dies (cache.py:51, cache.py:62), so it only trips on real failures. test_failing_group_serves_its_last_values (tests/test_exporter.py:73) would become the within-bound case, with a new past-bound test asserting the family is absent.

2. The default PLATFORM_METRICS_QUEUES misses four compose queues

The comment at backend/backend/settings/base.py:1178-1181 says the default is what the compose consumers read. But the notification consumer also reads notifications_webhook, notifications_email, notifications_sms and notifications_priority (docker/docker-compose.yaml:594), and the list at base.py:1182-1212 doesn't include them. An unlisted queue is exported only while it holds messages (groups.py:127), so when it drains it disappears instead of reading 0. A KEDA trigger on it then errors every time the queue is idle. Could you add those four? Cloud will override the list for its own extra queues anyway.

3. Smaller: the refresh runs inside whichever scrape takes the lock

The refresh runs in the request thread (cache.py:173 → groups.py:70-75), and every scrape resolves all three groups (exporter.py:57-59). The executions refresh can take up to 2 × 5 s (groups.py:170), while KEDA's HTTP client gives up after 3 s by default. A slow executions refresh therefore fails the KEDA request that happened to take that lock, even though KEDA only needs the queue series.

A failed refresh also isn't backed off. The next scrape that finds the snapshot stale tries again (cache.py:140-173), and at KEDA's polling rate that's several attempts a second against a struggling DB.

Neither blocks this PR. A ?group=queues filter, plus recording the failed attempt so retries stay at about one per TTL, would cover both, and this could also wait for the autoscaling ticket.

4. FYI, no change needed: claimed includes lapsed leases

claimed counts every state='claimed' row. That includes leases that have expired but that the reaper hasn't re-armed yet (workers/queue_backend/pg_queue/reaper.py:351-377), fire-and-forget failures left for redelivery (consumer.py:672-681), and poison re-parks (consumer.py:989). For scaling on ready + claimed that's fine, because that work comes back. The exception is when the reaper isn't running: expired claims then pile up as demand that no worker can claim. A separate unstract_queue_claims_expired{queue} would make that visible, but it's optional.

The rest already fails closed, which is what KEDA needs:

• known queues are zero-filled (groups.py:141-145);
• a cold start, with no snapshot and the lock held elsewhere, leaves the group out (cache.py:155-157, exporter.py:67-68);
• a snapshot that won't render is dropped (exporter.py:69-78).

Nice work on those.

[muhammad-ali-e]

@ritwik-g thanks, that's a useful review. Taking each point in turn, all addressed in 6d6c501:

1. Stale queue counts: done, as you proposed. MetricGroup.max_staleness_seconds defaults to None. QueueMetricGroup reads it from PLATFORM_METRICS_QUEUE_MAX_STALENESS_SECONDS (default 60). Past that bound, collect() skips the group's families but still emits its unstract_metrics_last_successful_refresh_timestamp sample and returns 200, and the other groups are unaffected. test_failing_group_serves_its_last_values still covers the within-bound case. The new test_group_past_its_staleness_bound_is_left_out covers both sides of the bound and checks that an unbounded group keeps serving. Executions and batches stay unbounded because only autoscaling reads them in a way where a frozen value causes harm. That's easy to change if you'd rather bound them too.

2. Notification sub-queues: added notifications_email, notifications_priority, notifications_sms and notifications_webhook to the default list.

3. Done here rather than deferred, since both changes were small:

• ?group=queues (repeatable) renders only the selected groups, so a KEDA poll never waits on the executions refresh. An unselected group isn't refreshed at all, and an unknown name returns 400 so a typo in a trigger fails loudly. Each render builds its own CollectorRegistry from the selected groups.
• Backoff: a failed refresh now keeps the Redis lock and shortens its TTL to the cache TTL (Lock.extend(ttl, replace_ttl=True)) instead of releasing it. Every pod serves the last snapshot until it lapses, so retries stay at about one per TTL across the cluster, whatever the scrape rate. The Redis-less path does the same per process by recording its last attempt. Both are covered by new tests.

4. Lapsed claims: agreed, not in this PR. Ready + claimed is the right scaling signal while the reaper runs. unstract_queue_claims_expired{queue} fits better with the reaper-health work, so I'll note it on UN-4525.

51 tests pass locally. CI is re-running.

---

Generated by Claude Code

[sonarqubecloud]

Quality Gate passed

Issues
2 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[github-actions]

Unstract test results

Per-group results

Status	Group	Tier	Passed	Failed	Errors	Skipped	Duration (s)
⚪	e2e-agent-kv	e2e	0	0	0	1	0.1
✅	e2e-api-deployment	e2e	3	0	0	0	12.2
✅	e2e-coowners	e2e	1	0	0	0	1.3
✅	e2e-etl	e2e	1	0	0	0	14.2
✅	e2e-login	e2e	2	0	0	0	1.4
✅	e2e-prompt-studio	e2e	1	0	0	0	5.2
✅	e2e-smoke	e2e	2	0	0	0	1.3
✅	e2e-workflow	e2e	1	0	0	0	16.2
✅	frontend	unit	629	0	0	0	13.9
✅	integration-backend	integration	1453	0	0	26	86.4
✅	integration-connectors	integration	1	0	0	7	9.6
✅	integration-workers	integration	181	0	0	1	56.4
❌	ui	e2e	0	1	0	0	0.0
✅	unit-agent-kv-schema	unit	143	0	0	0	1.3
✅	unit-backend	unit	2098	0	0	1	45.6
✅	unit-connectors	unit	82	0	0	0	8.7
✅	unit-core	unit	281	0	0	0	2.3
✅	unit-filesystem	unit	1	0	0	0	3.2
✅	unit-platform-service	unit	18	0	0	0	2.0
✅	unit-rig	unit	120	0	0	0	3.4
✅	unit-runner	unit	12	0	0	0	2.3
✅	unit-sdk1	unit	878	0	0	0	33.6
✅	unit-tool-sidecar	unit	8	0	0	0	1.3
✅	unit-workers	unit	1534	0	0	2	117.3
	TOTAL		7450	1	0	38	439.2

Critical paths

⚠️ Critical paths not yet covered

• workflow-execution-fan-out — Multi-file workflow execution fans out to file-processing workers and rejoins. (declared coverage: no groups declared)

✅ Covered critical paths

• auth-login — covered by e2e-login
• adapter-register-llm — covered by integration-backend
• workflow-author — covered by integration-backend
• co-owner-manage — covered by integration-backend, e2e-coowners
• workflow-create-execute — covered by e2e-workflow
• api-deployment-provision — covered by integration-backend
• api-deployment-auth — covered by integration-backend
• api-deployment-run — covered by e2e-api-deployment
• mcp-server-auth — covered by integration-backend
• mcp-platform-auth — covered by integration-backend
• platform-key-whoami — covered by integration-backend
• prompt-studio-author — covered by integration-backend
• prompt-studio-fetch-response — covered by e2e-prompt-studio
• connector-register-test — covered by integration-backend
• pipeline-etl-execute — covered by e2e-etl
• usage-aggregate-read — covered by integration-backend
• usage-token-tracking — covered by e2e-api-deployment
• callback-result-delivery — covered by e2e-api-deployment