xsl
746b66da01
One request exercises every layer: JWT auth, quota reservation, a versioned prompt through the LLM gateway, a cost row, quota commit or refund, and the unified envelope. The trivial ping task proves the queue routing and the tk_task transitions before any real long-running task is written. Routes are mounted at the /api/v1 base path of 架构文档 §9.1; /v1/* is rewritten to it at the ASGI layer so both spellings work from one route table.
60 lines
2.2 KiB
Python
Executable File
60 lines
2.2 KiB
Python
Executable File
"""Liveness, readiness and metrics. Mounted at the root, outside ``/api/v1``.
|
|
|
|
``/healthz`` must never touch a dependency: if it did, a slow database would
|
|
make the orchestrator kill healthy processes. ``/readyz`` is the one that checks
|
|
dependencies, because a process that cannot reach PostgreSQL should be taken out
|
|
of the load-balancer rotation but not restarted.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Response
|
|
from prometheus_client import CONTENT_TYPE_LATEST, generate_latest
|
|
from sqlalchemy import text
|
|
|
|
from packages.shared.config import get_settings
|
|
from packages.shared.db import get_session_factory
|
|
from packages.shared.errors import AppError, ErrorCode
|
|
from packages.shared.logging import get_logger
|
|
from packages.shared.observability import REGISTRY
|
|
from packages.shared.redis_client import get_redis
|
|
from packages.shared.response import ok
|
|
|
|
router = APIRouter(tags=["ops"])
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
@router.get("/healthz", summary="Liveness probe")
|
|
async def healthz() -> dict[str, object]:
|
|
settings = get_settings()
|
|
return ok({"status": "ok", "app": settings.app_name, "env": settings.app_env})
|
|
|
|
|
|
@router.get("/readyz", summary="Readiness probe (checks PostgreSQL and Redis)")
|
|
async def readyz() -> dict[str, object]:
|
|
checks: dict[str, str] = {}
|
|
|
|
try:
|
|
factory = get_session_factory()
|
|
async with factory() as session:
|
|
await session.execute(text("SELECT 1"))
|
|
checks["postgres"] = "ok"
|
|
except Exception as exc:
|
|
checks["postgres"] = f"error: {type(exc).__name__}"
|
|
|
|
try:
|
|
await get_redis().ping()
|
|
checks["redis"] = "ok"
|
|
except Exception as exc:
|
|
checks["redis"] = f"error: {type(exc).__name__}"
|
|
|
|
if any(v != "ok" for v in checks.values()):
|
|
logger.warning("readiness_failed", checks=checks)
|
|
raise AppError(ErrorCode.UPSTREAM_ERROR, message="依赖未就绪", data={"checks": checks})
|
|
return ok({"status": "ready", "checks": checks})
|
|
|
|
|
|
@router.get("/metrics", summary="Prometheus scrape endpoint", include_in_schema=False)
|
|
async def metrics() -> Response:
|
|
return Response(content=generate_latest(REGISTRY), media_type=CONTENT_TYPE_LATEST)
|