Health checks в FastAPI: liveness, readiness и /info

← назад к разделу

Kubernetes, балансировщики и мониторинг постоянно спрашивают каждый сервис: «ты жив?» и «ты готов принимать запросы?». Если сервис отвечает неправильно — трафик пойдёт не туда, поды начнут перезапускаться, и одна небольшая проблема с базой данных превратится в полный отказ сервиса.

Разберём, как устроить эти проверки правильно.

Зачем нужны два разных эндпоинта

Кажется логичным: есть один /health, он отвечает «ок» или «не ок». Но это создаёт серьёзную проблему.

Kubernetes использует два разных вида проверок:

Liveness probe — проверяет, живой ли процесс. Если нет — Kubernetes убивает под и запускает новый.
Readiness probe — проверяет, готов ли сервис принимать трафик. Если нет — убирает под из балансировки, но не перезапускает.

Если смешать обе проверки в одну и добавить туда проверку базы данных, получится катастрофа: база притормозила на 30 секунд из-за планового обслуживания → liveness возвращает ошибку → Kubernetes убивает все поды → новые поды стартуют с той же проблемной базой → снова падают → цикл. Сервис полностью недоступен из-за временного лага в БД.

Правильное решение: разделить.

from fastapi import FastAPI

app = FastAPI()


@app.get("/health/live")
async def liveness() -> dict:
    return {"status": "UP"}


@app.get("/health/ready")
async def readiness() -> dict:
    checks = await run_readiness_checks()
    status = "UP" if all(c["status"] == "UP" for c in checks) else "DOWN"
    return {"status": status, "checks": checks}

Эндпоинт	Что проверяет	Что делает Kubernetes
`/health/live`	Процесс отвечает, event loop не завис	UP → продолжать; DOWN → перезапускает под
`/health/ready`	БД доступна, зависимости готовы	UP → шлёт трафик; DOWN → убирает из балансировки

Семантика: если база недоступна 5 секунд — readiness должен стать DOWN (трафик уйдёт на другие реплики), а liveness должен остаться UP (перезапуск пода не поможет, база та же самая).

Liveness без внешних зависимостей

Liveness должна проверять только сам процесс — жив ли event loop, отвечает ли приложение. Никаких внешних систем.

Вот пример того, как делать не надо:

# Опасно: liveness вернёт 503 при любой недоступности БД
@app.get("/health/live")
async def liveness_bad():
    result = await check_postgres(postgres_pool)
    if result["status"] != "UP":
        raise HTTPException(status_code=503, detail=result)
    return {"status": "UP"}

При таком подходе временный лаг Postgres запустит перезапуск всех подов. Правильно — только проверка самого приложения:

@app.get("/health/live")
async def liveness() -> dict:
    return {"status": "UP"}

Проверка внешних систем с кешем

Readiness проверяет реальные зависимости: базу данных, Redis, внешние API. Но здесь есть другая ловушка: Kubernetes опрашивает readiness каждые 5 секунд, и если у вас 10 реплик — это 120 запросов в минуту только от health-check. Для дорогих внешних проверок это много.

Решение — TTL-кеш: результат проверки сохраняется на несколько секунд, и повторные запросы берут его из кеша.

import asyncio
import time
from dataclasses import dataclass
from typing import Optional

import httpx


@dataclass
class CachedCheck:
    result: dict
    expires_at: float


_payment_cache: Optional[CachedCheck] = None
_payment_lock = asyncio.Lock()
TTL_SECONDS = 10


async def check_payment_api() -> dict:
    global _payment_cache
    now = time.monotonic()

    if _payment_cache and _payment_cache.expires_at > now:
        return _payment_cache.result

    async with _payment_lock:
        if _payment_cache and _payment_cache.expires_at > now:
            return _payment_cache.result

        result = await _do_check_payment()
        _payment_cache = CachedCheck(result=result, expires_at=now + TTL_SECONDS)
        return result


async def _do_check_payment() -> dict:
    try:
        async with httpx.AsyncClient(timeout=2.0) as client:
            resp = await client.get("https://api.payment-provider.ru/ping")
            resp.raise_for_status()
        return {"status": "UP", "component": "payment"}
    except Exception as exc:
        return {"status": "DOWN", "component": "payment", "error": str(exc)}

Двойная проверка внутри asyncio.Lock защищает от состояния гонки: если несколько корутин одновременно обнаружили устаревший кеш, только одна пойдёт к внешнему API, остальные дождутся и возьмут свежий результат.

Проверки для PostgreSQL и Redis проще — у них уже есть нативные ping-методы:

async def check_postgres(pool: asyncpg.Pool) -> dict:
    try:
        async with pool.acquire(timeout=1.0) as conn:
            await conn.fetchval("SELECT 1")
        return {"status": "UP", "component": "postgres"}
    except Exception as exc:
        return {"status": "DOWN", "component": "postgres", "error": str(exc)}


async def check_redis(redis_client) -> dict:
    try:
        await redis_client.ping()
        return {"status": "UP", "component": "redis"}
    except Exception as exc:
        return {"status": "DOWN", "component": "redis", "error": str(exc)}

Собираем итоговый readiness-эндпоинт:

from fastapi import APIRouter, Response

health_router = APIRouter()


@health_router.get("/health/ready")
async def readiness(response: Response) -> dict:
    checks = await asyncio.gather(
        check_postgres(postgres_pool),
        check_redis(redis_client),
        check_payment_api(),
    )
    checks_list = list(checks)
    all_up = all(c["status"] == "UP" for c in checks_list)

    if not all_up:
        response.status_code = 503

    return {
        "status": "UP" if all_up else "DOWN",
        "checks": checks_list,
    }

HTTP-статус 503 при DOWN обязателен: Kubernetes читает именно код ответа, а не содержимое тела.

Эндпоинт /info с версией и git-коммитом

Во время инцидента первый вопрос — «какая версия сервиса сейчас в продакшне?». Если специального эндпоинта нет, придётся лезть в CI-логи и сопоставлять даты деплоев.

Решение: при сборке записать метаданные в файл, а FastAPI отдаёт их по запросу.

В CI-пайплайне:

# .github/workflows/build.yml
- name: Write build info
  run: |
    cat > app/build_info.json << EOF
    {
      "git_commit": "${{ github.sha }}",
      "git_branch": "${{ github.ref_name }}",
      "build_time": "$(date -u +%Y-%m-%dT%H:%M:%SZ)",
      "version": "${{ github.run_number }}"
    }
    EOF

В приложении:

import json
from pathlib import Path

from fastapi import APIRouter

info_router = APIRouter()


def _load_build_info() -> dict:
    path = Path(__file__).parent / "build_info.json"
    if path.exists():
        return json.loads(path.read_text())
    return {"git_commit": "dev", "version": "local"}


_BUILD_INFO = _load_build_info()


@info_router.get("/info")
async def info() -> dict:
    return {
        "service": {"name": "order-service"},
        "build": _BUILD_INFO,
    }

Ответ выглядит так:

{
  "service": {"name": "order-service"},
  "build": {
    "git_commit": "5380f21abc...",
    "git_branch": "main",
    "build_time": "2026-06-18T14:30:00Z",
    "version": "142"
  }
}

Отдельный порт для management-эндпоинтов

Health-эндпоинты лучше держать на отдельном порту — чтобы они не были доступны через публичный балансировщик и не мешали основному трафику.

import asyncio
import uvicorn
from fastapi import FastAPI

management_app = FastAPI(docs_url=None, redoc_url=None)
management_app.include_router(health_router)
management_app.include_router(info_router)


async def serve_both():
    config_business = uvicorn.Config(app, host="0.0.0.0", port=8080)
    config_management = uvicorn.Config(management_app, host="0.0.0.0", port=9090)
    await asyncio.gather(
        uvicorn.Server(config_business).serve(),
        uvicorn.Server(config_management).serve(),
    )

asyncio.run(serve_both())

В Kubernetes-манифесте указываем порт 9090 для проб:

spec:
  containers:
    - name: order-service
      livenessProbe:
        httpGet:
          path: /health/live
          port: 9090
        initialDelaySeconds: 10
        periodSeconds: 10
        failureThreshold: 3
      readinessProbe:
        httpGet:
          path: /health/ready
          port: 9090
        initialDelaySeconds: 5
        periodSeconds: 5
        failureThreshold: 2

Частые ошибки

Бизнес-логика в health-check. Если сделать if order_count > N: status = "DOWN" — Kubernetes начнёт перезапускать сервис из-за бизнес-состояния. Health check должен отражать только техническое состояние: процесс жив, база доступна.

Тяжёлые операции как проба. Создавать тестовый заказ или выполнять полный запрос как часть probe — плохая идея. Probe должна быть лёгкой: ping, SELECT 1, простой GET-запрос.

Readiness всегда 200, но DOWN в теле. Kubernetes смотрит на HTTP-статус, не на тело. Если readiness возвращает 200 при DOWN — Kubernetes думает, что всё хорошо, и продолжает слать трафик.

Проверки без кеша. Без TTL-кеша каждый опрос Kubernetes идёт напрямую к внешней системе. При частом опросе и нескольких репликах это ощутимая нагрузка.

Коротко

/health/live и /health/ready — два разных эндпоинта с разной семантикой. Смешивать нельзя.
Liveness проверяет только сам процесс. Внешние системы туда не включают.
Readiness проверяет все зависимости: БД, Redis, внешние API.
Для внешних API-проверок используйте TTL-кеш с asyncio.Lock — иначе health-check создаёт лишнюю нагрузку.
HTTP 503 при DOWN в readiness обязателен — Kubernetes читает статус, не тело.
/info с git-коммитом и версией сборки сильно упрощает разбор инцидентов.
Management-эндпоинты лучше держать на отдельном порту.

Что почитать дальше

Logging — structlog, contextvars и structured logging
Metrics — prometheus-client и бизнес-метрики
Tracing — OpenTelemetry и ручные spans