---
title: 'TEXTIQ-391: Prometheus /metrics on DE Airflow plugin'
type: 'feature'
created: '2026-05-04'
status: 'complete'
completed: '2026-05-04'
baseline_commit: dc48a00
final_commit: 32d68e7
scope_note: 'G2 (StatsD exporter sidecar) descoped — deferred to GAP-029 Phase 2. Plugin /metrics (G1) and ops scrape doc (G3) delivered.'
auth_note: '/metrics now requires HMAC (post-completion hardening, 2026-05-04). Internal TextIQ services scrape with the same signing pattern as other endpoints; vanilla Prometheus needs a signing proxy.'
context:
  - '_planning/docs/planning-artifacts/2-planning/system-gaps-and-changes.md (GAP-029)'
  - '_planning/docs/services/textiq-doc-extraction/architecture.md (Section 6.2)'
---

<frozen-after-approval reason="human-owned intent — do not modify unless human renegotiates">

## Intent

**Problem:** DE Airflow plugin has no Prometheus instrumentation. SYS-NFR-33 monitoring stack is non-operational for the document-extraction service: no `/metrics` endpoint, no `prometheus_client` dependency, no Airflow task-level metrics emitted. Ops cannot observe upload throughput, approval/rejection rates, HMAC auth latency, or DAG task health.

**Approach:** Add `prometheus-client` to the api-server runtime, expose `/metrics` on each FastAPI sub-app (`upload`, `tags`, `domain_terms`), instrument the four upload-flow counters + an upload-duration histogram + an HMAC-verify-duration histogram + at least one counter/histogram per remaining sub-app. For Airflow task-level metrics, enable Airflow's built-in StatsD emitter and add a `prom/statsd-exporter` sidecar so Prometheus can scrape Airflow internals on `:9102`. Document the scrape config for ops.

## Boundaries & Constraints

**Always:**
- Use `prometheus_client`'s default global registry. Pin api-server to a single worker via `AIRFLOW__API__WORKERS=1` — single-node / single-worker is the current deployment shape; multi-worker scraping (multiprocess collector) is explicitly out of scope until that changes.
- Bounded-cardinality labels only: `outcome ∈ {success, client_error, server_error}`, `tenant_id` (UUID), `operation` where applicable. Derive `outcome` from the response's final `status_code`, not from try/except control flow.
- Install `prometheus-client` into BOTH the venv (`uv add prometheus-client`) AND Airflow's own Python via Dockerfile pip install — same dual-install pattern as `pymilvus` (Dockerfile.airflow:51).
- Plugin code stays import-pure: stdlib + airflow-runtime-installed packages only. Do NOT import from `src.*` (per existing module docstrings).

**Ask First:**
- If the dev-mount diff already on `docker-compose.airflow.yml` (carried from `dev`) conflicts with the new sidecar/env additions, HALT and confirm merge order.
- If `prom/statsd-exporter` mapping rules need to grow beyond a minimal seed set, surface for review rather than guessing.
- If api-server worker count ever needs to be raised above 1, HALT — counter/histogram registry must be migrated to `MultiProcessCollector` first (separate spec).

**Never:**
- No OpenTelemetry, Loki, Grafana dashboards, or alert rules — those are GAP-029 Phase 2, out of scope.
- No `user_id`, `correlation_id`, or `request_id` labels — unbounded cardinality. Use logs/exemplars instead.
- No behaviour change to existing endpoints — instrumentation must be observation-only.
- No commented-out stub style (the IAM `core/metrics.py` anti-pattern). Register live metrics from day one.

## I/O & Edge-Case Matrix

| Scenario | Input / State | Expected Output / Behavior | Error Handling |
|---|---|---|---|
| Successful upload | `POST /api/v1/documents/upload` returns 202 | `upload_total{outcome="success",tenant_id=<uuid>}` +1; `upload_duration_seconds` observation recorded | N/A |
| Validation-failed upload | `POST .../upload` returns 4xx (bad ext / oversize / password) | `upload_total{outcome="client_error",tenant_id=<uuid>}` +1; duration still recorded | Counter increments before response returns |
| HMAC failure | Request returns 401 from middleware | `upload_total` NOT incremented (route never enters); `hmac_verify_duration_seconds` IS recorded with `result="fail"` label | tenant_id unknown at this stage; omit label |
| `/metrics` per sub-app | `GET .../documents/metrics`, `.../blocks/metrics`, `.../domain-terms/metrics` | All three return identical Prometheus payload (shared global registry, single worker); scrape config picks one | N/A |
| StatsD exporter down | sidecar unavailable | Airflow continues to emit (UDP); samples dropped; Prometheus scrape of `:9102` fails | Non-fatal to extraction pipeline; documented in runbook |

</frozen-after-approval>

## Code Map

- `pyproject.toml` + `uv.lock` -- managed via `uv add prometheus-client`; do NOT hand-edit `pyproject.toml`
- `docker/Dockerfile.airflow:48-52` -- pip-install `prometheus-client` into Airflow's own Python (mirror line 51 `pymilvus` pattern)
- `docker/docker-compose.airflow.yml` -- add `airflow-statsd-exporter` service; add Airflow `[metrics]` StatsD env vars + `AIRFLOW__API__WORKERS=1` to `x-airflow-env`
- `docker/statsd-mapping.yml` -- new; statsd-exporter mapping rules for Airflow internal metrics
- `airflow/plugins/api/metrics.py` -- new; declare counters/histograms against the default global registry, `metrics_response()` helper, `register_metrics_route(app)`, `record_outcome(counter, status_code, **labels)` helper
- `airflow/plugins/api/middleware.py:63-97` -- time HMAC verify with `hmac_verify_duration_seconds.labels(result=...).time()`
- `airflow/plugins/api/upload.py:81,247,314,401` -- record `upload/approve/reject/delete_total{outcome,tenant_id}` + `upload_duration_seconds`
- `airflow/plugins/api/tags.py:150,195,281,336` -- `tag_op_total{outcome,operation,tenant_id}` + `tag_endpoint_duration_seconds{operation}`
- `airflow/plugins/api/domain_terms.py:123,190,276,332` -- `domain_term_op_total{outcome,operation,tenant_id}` + `domain_term_endpoint_duration_seconds{operation}`
- `airflow/plugins/__init__.py:21-44` -- call `register_metrics_route(app)` for each of the three sub-apps
- `docs/runbooks/prometheus-scraping.md` -- new; scrape-target snippet (DE plugin + statsd-exporter)

## Tasks & Acceptance

**Execution:**
- [x] `uv add prometheus-client` -- adds unpinned dep to `pyproject.toml` + `uv.lock` for venv-side availability (tests / future DAG-task instrumentation); do NOT hand-edit `pyproject.toml`
- [x] `docker/Dockerfile.airflow` -- mirror line 51 pattern: `pip install prometheus-client` into Airflow's own Python (unpinned)
- [x] `airflow/plugins/api/metrics.py` -- new module: declare all counters/histograms (see Design Notes) on the default global registry; `metrics_response()` returns `generate_latest()` with `CONTENT_TYPE_LATEST`; `register_metrics_route(app)` mounts `/metrics` on the given sub-app; `record_outcome(counter, status_code, **labels)` helper maps status_code → outcome label
- [x] `airflow/plugins/api/middleware.py` -- wrap HMAC verify call site with `hmac_verify_duration_seconds.labels(result=...).time()`; record on both success and 401/500 paths
- [x] `airflow/plugins/api/upload.py` -- instrument `upload`, `delete_document`, `approve_document`, `reject_document`; tenant_id from `request.state.hmac_ctx["tenant_id"]`
- [x] `airflow/plugins/api/tags.py` -- instrument the four mutating endpoints with shared tag counter + duration histogram
- [x] `airflow/plugins/api/domain_terms.py` -- instrument the four mutating endpoints with shared domain-term counter + duration histogram
- [x] `airflow/plugins/__init__.py` -- mount `/metrics` on each sub-app via `register_metrics_route`
- [x] `docker/docker-compose.airflow.yml` -- add `AIRFLOW__API__WORKERS=1` to `x-airflow-env`
- [~] ~~`docker/docker-compose.airflow.yml` -- add `airflow-statsd-exporter` sidecar + `AIRFLOW__METRICS__STATSD_*` env vars~~ -- **DESCOPED** (2026-05-04): Airflow task-level metrics (G2) deferred to GAP-029 Phase 2. GAP-029 Phase 1 only requires plugin `/metrics`; ticket AC verifies plugin only. See ticket comment for rationale.
- [~] ~~`docker/statsd-mapping.yml`~~ -- **DESCOPED**: removed alongside the sidecar.
- [x] `tests/airflow/test_metrics.py` -- new; cover all I/O matrix rows (success, client_error, HMAC failure, /metrics shape per sub-app)
- [x] `docs/runbooks/prometheus-scraping.md` -- new; two scrape jobs (DE plugin → `/api/v1/documents/metrics`, statsd-exporter → `:9102/metrics`)

**Acceptance Criteria:**
- Given api-server is up, when `curl .../api/v1/documents/metrics`, then `Content-Type: text/plain; version=0.0.4` and body contains `# TYPE upload_total counter` plus `# TYPE upload_duration_seconds histogram`.
- Given the same scrape against `.../api/v1/blocks/metrics` and `.../api/v1/domain-terms/metrics`, then each response includes ≥1 counter + ≥1 histogram registered for that sub-app's domain (per-sub-app AC).
- Given a successful upload, when `/metrics` is scraped, then `upload_total{outcome="success",tenant_id=<uuid>}` shows the increment.
- Given a 4xx upload (e.g. PASSWORD_PROTECTED), when `/metrics` is scraped, then `upload_total{outcome="client_error",tenant_id=<uuid>}` is incremented (not `success`).
- Given the api-server runs with `AIRFLOW__API__WORKERS=1`, when concurrent uploads hit the single worker, then `/metrics` reflects all increments correctly (no per-worker drift to worry about).
- Given the statsd-exporter sidecar is running and a DAG completes, when `curl statsd-exporter:9102/metrics`, then at least one `airflow_*` series appears for that `dag_id`.
- Given `docs/runbooks/prometheus-scraping.md` exists, when an ops engineer reads it, then they can configure both scrape jobs without consulting source code.

## Design Notes

**Counter / histogram inventory** (declared at module scope in `metrics.py`):
- Counters: `upload_total`, `approve_total`, `reject_total`, `delete_total` — labels `(outcome, tenant_id)`
- Counters: `tag_op_total`, `domain_term_op_total` — labels `(outcome, operation, tenant_id)`
- Histograms: `upload_duration_seconds` — labels `(outcome,)`; `hmac_verify_duration_seconds` — labels `(result,)`; `tag_endpoint_duration_seconds`, `domain_term_endpoint_duration_seconds` — labels `(operation,)`
- Buckets: `prometheus_client` defaults for HMAC; explicit `(0.1, 0.5, 1, 2, 5, 10, 30, 60, 120)` seconds for upload (large files dominate tail).

**`/metrics` handler** (single-worker, default global registry):
```python
def metrics_response():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)
```
If api-server is ever scaled to workers > 1, this must switch to `MultiProcessCollector` + a writable `PROMETHEUS_MULTIPROC_DIR` — explicitly out of scope here, gated by the **Ask First** rule.

**Outcome derivation:** record at end of handler from the actual response's `status_code` so handler-thinks-success-but-returns-500 paths still report correctly:
```python
resp = success_response({...}, status_code=202)
record_outcome(upload_total, resp.status_code, tenant_id=tenant_id)
upload_duration_seconds.labels(outcome=outcome_for(resp.status_code)).observe(elapsed)
return resp
```

**Cardinality:** `tenant_id` UUID × 3 outcomes × 4 upload counters ≈ 12 series per tenant. At 100 tenants → ~1200 series for upload-flow. Acceptable. Revisit if tenant count > 1000 (option: drop tenant_id from histograms).

**StatsD mapping:** seed with three rules covering DAG-run duration + task start/finish; let ops extend in PRs as needs surface. `prom/statsd-exporter` README has Airflow-shaped examples.

## Verification

**Commands:**
- `docker compose -f docker/docker-compose.airflow.yml up -d airflow-init airflow-webserver airflow-scheduler airflow-statsd-exporter` -- expected: all containers reach `healthy`.
- `curl -s http://localhost:8081/api/v1/documents/metrics | grep -E '^# TYPE (upload_total|upload_duration_seconds|hmac_verify_duration_seconds)'` -- expected: 3 matching lines.
- `curl -s http://localhost:8081/api/v1/blocks/metrics | grep -c '^# TYPE'` and same for `/domain-terms/metrics` -- expected: ≥2 each (counter + histogram).
- `curl -s http://localhost:9102/metrics | grep -c '^airflow_'` -- expected: ≥1 after triggering a DAG.
- `pytest tests/airflow/test_metrics.py -v` -- expected: all matrix scenarios pass.
- `docker compose exec airflow-webserver python -c "import prometheus_client; print(prometheus_client.__version__)"` -- expected: prints version, no ImportError (proves dual install worked).

**Manual checks:**
- After one successful upload + one bad-extension upload, `/metrics` shows `upload_total{outcome="success"}` = 1 and `upload_total{outcome="client_error"}` = 1.
- DAG run for `document_extraction_dag` produces non-empty `airflow_*` series on statsd-exporter within 30 s of completion.

## Post-Completion Changes

### 2026-05-04 — Enable HMAC auth on `/metrics`

**Decision:** removed the `/metrics` path exemption from `HMACVerificationMiddleware` so the endpoint is auth-protected like every other plugin route.

**Rationale:** the original implementation (commit `ef50c2d`) bypassed HMAC for `/metrics` so vanilla Prometheus could scrape unauthenticated. After review with the user, we accept that scrapers in our trust boundary are TextIQ services holding the shared key ring, not vanilla Prometheus — they sign `/metrics` exactly like upload/approve/etc. If a vanilla Prometheus scraper is ever needed, a signing sidecar/proxy stays as the documented option (it is NOT in scope here).

**Behavior:**
- Unsigned `GET .../metrics` → `401 UNAUTHORIZED` (`Missing header: X-Node-ID`).
- Signed scrape (textiq_hmac headers) → `200 OK` with the standard Prometheus text payload.
- Each scrape now contributes to `hmac_verify_duration_seconds{result="success"|"fail"}` like any other endpoint — observability of the observability endpoint.

**Verification (executed):**
- Unsigned scrape returned 401.
- Signed scrape returned 200; full 5,501-byte payload contained the expected `# TYPE upload_total counter` etc.
- Signed upload + signed approve still pass; counters reflect both ops; HMAC histogram reflects scrapes.

**Files touched:** `airflow/plugins/api/middleware.py` (4-line removal of the path-suffix exemption).

**Trade-off accepted:** vanilla Prometheus cannot scrape this endpoint without a signing intermediary — documented in the runbook so ops can plan accordingly.