Observability, metrics, and Flower (FastAPI + Celery)

Audience: Teams shipping FastAPI + Celery + RabbitMQ who need production visibility. Golden rule: Flower is an ops tool. Product job status always comes from your application `jobs` table (or equivalent), never from Flower or the broker UI.

TL;DR

Client ──► FastAPI ──► jobs row + publish ──► RabbitMQ ──► workers
 │ │
 │ logs/metrics │ logs/metrics + Flower (ops)
 ▼ ▼
 GET /jobs/{id} ◄── product truth ── application DB

Contents

1. API health endpoints
2. Structured logging and correlation
3. Core job metrics
4. SLOs
5. Celery + Flower setup
6. Securing admin UIs
7. What to alert on
8. Worker and API instrumentation pseudocode
9. Dashboard sketch
10. Anti-patterns
11. Checklist

---

1. API health endpoints

Usually do not make API readiness depend on “workers are healthy.” Scale and restart workers independently. Optionally expose a worker readiness that checks broker connectivity for the worker deployment only.

# PSEUDOCODE
@app.get("/health")
async def health():
 return {"status": "ok"}

@app.get("/ready")
async def ready(db: Db = Depends()):
 await db.execute("SELECT 1")
 return {"status": "ready"}

---

2. Structured logging and correlation

Use JSON logs to stdout (platform scrapes them).

Fields to include

# PSEUDOCODE : API enqueue
log.info(
 "job_enqueued",
 extra={
 "job_id": str(job.id),
 "job_type": job.type,
 "queue": "jobs.io",
 "request_id": request_id,
 },
)

# PSEUDOCODE : worker
log.info(
 "job_started",
 extra={"job_id": job_id, "task_id": self.request.id, "attempts": self.request.retries},
)

Propagate ids: HTTP header X-Request-Id → log context → message headers → worker logging context.

Optional: OpenTelemetry trace context in message headers for API → worker spans.

---

3. Core job metrics

Track at least:

Success rate ≈ succeeded / (succeeded + failed) over a window (exclude still-running).

Backlog = queue depth (and/or lag if you also use Kafka). Export via RabbitMQ exporter or management API scraper.

---

4. SLOs

Define per critical job type (not one global number):

Measure from the `jobs` table timestamps so product and ops share the same truth.

time_to_start = started_at - created_at
time_to_complete = finished_at - created_at

---

5. Celery + Flower setup

Workers and Beat (reminder)

celery -A app.workers.celery_app.celery_app worker -Q jobs.default,jobs.io -c 4
celery -A app.workers.celery_app.celery_app beat -l info # single leader

Flower

# PSEUDOCODE : never expose publicly without auth
celery -A app.workers.celery_app.celery_app flower \
 --port=5555 \
 --basic_auth=ops_user:strong_password

Compose sketch:

# PSEUDOCODE
services:
 flower:
 image: your-app:tag
 command: celery -A app.workers.celery_app.celery_app flower --port=5555
 # no public ports in prod; private network only
 environment:
 CELERY_BROKER_URL: ${CELERY_BROKER_URL}

What Flower is good for

• See active tasks, workers online, basic task history
• Debug “is any worker connected?”
• Inspect failures during incidents

What Flower is not

• Not the customer “is my export done?” API
• Not the source of truth for SLAs (use DB timestamps + metrics)
• Not a substitute for RabbitMQ queue depth / DLQ alerts

---

6. Securing admin UIs

Mandatory for production:

Checklist:

• [ ] No public 0.0.0.0 without auth
• [ ] SSO or strong basic auth + network policy
• [ ] TLS terminate at reverse proxy
• [ ] Separate credentials from app DB users
• [ ] Audit who can access ops UIs

---

7. What to alert on

Page on Critical/High; ticket or daytime on Medium unless prolonged.

---

8. Worker and API instrumentation pseudocode

# PSEUDOCODE : enqueue
async def enqueue(...):
 job = await db.insert_job(...)
 await bus.publish(queue, {"job_id": str(job.id)})
 metrics.incr("jobs_enqueued_total", tags={"type": job.type, "queue": queue})
 log.info("job_enqueued", extra={"job_id": str(job.id), "request_id": rid})
 return job

# PSEUDOCODE : worker
@celery_app.task(bind=True)
def send_receipt(self, job_id: str):
 t0 = time.perf_counter()
 log.info("job_started", extra={"job_id": job_id, "task_id": self.request.id})
 try:
 mark_running(job_id)
 do_work(job_id)
 mark_succeeded(job_id)
 metrics.incr("jobs_succeeded_total", tags={"type": "send_receipt"})
 except TransientError:
 metrics.incr("jobs_retries_total", tags={"type": "send_receipt"})
 raise
 except Exception:
 mark_failed(job_id, ...)
 metrics.incr("jobs_failed_total", tags={"type": "send_receipt"})
 raise
 finally:
 metrics.observe("job_runtime_seconds", time.perf_counter() - t0, tags={"type": "send_receipt"})

Export Prometheus (or your vendor) from API and workers; scrape RabbitMQ exporter for depth/consumers.

---

9. Dashboard sketch

Row 1 : Traffic: enqueue rate, success rate, retry rate Row 2 : Latency: time-to-start p95, time-to-complete p95 (by type) Row 3 : Broker: depth per queue, DLQ depth, consumer count Row 4 : Workers: process count, task runtime histogram Row 5 : Dependencies: DB errors, SMTP/HTTP client errors

Link runbooks: scale workers, redrive DLQ, restart Beat.

---

10. Anti-patterns

---

11. Checklist

Observability

• [ ] GET /health and GET /ready on API
• [ ] Structured JSON logs on API, worker, Beat
• [ ] request_id / job_id / event_id propagated end-to-end
• [ ] Counters: enqueue, success, fail, retries
• [ ] Histograms: runtime, time-to-start, time-to-complete
• [ ] Gauges: queue depth, DLQ depth, consumer count
• [ ] SLOs defined for critical job types

Flower and admin UIs

• [ ] Flower optional; command documented
• [ ] Flower on private network
• [ ] Authentication enabled
• [ ] TLS at the edge
• [ ] Same controls for RabbitMQ Management / Kafka UI
• [ ] Documented: Flower ≠ product status

Alerts

• [ ] Zero consumers → Critical
• [ ] DLQ depth > 0 (critical queues) → High
• [ ] Success rate drop → High
• [ ] Retry spike → Medium
• [ ] Time-to-complete SLO breach → Medium

---

Deep dive: 18 Monitoring Celery with Flower

Prometheus: 19 Integrate Flower with Prometheus

See also

• 05 Celery + RabbitMQ
• 09 Errors, retries, DLQ
• 14 Celery Beat
• 03 Job lifecycle (status in DB)
• 10 Testing

External

• Flower
• Celery monitoring
• OpenTelemetry