import asyncio import time import uuid from pathlib import Path import httpx from fastapi import FastAPI, Request from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, field_validator from opentelemetry import metrics, trace from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from opentelemetry.instrumentation.utils import suppress_instrumentation from common.chaos_flags import ALL_FLAGS, get_flags, set_flag, set_profile from common.logging_config import configure_logging from common.settings import settings from common.telemetry import setup_telemetry from . import orchestrator from .orchestrator import UpstreamFailure SERVICE_NAME = "gateway" logger = configure_logging(SERVICE_NAME) setup_telemetry(service_name=SERVICE_NAME) app = FastAPI(title=SERVICE_NAME) FastAPIInstrumentor.instrument_app( app, excluded_urls="/healthz,/api/health/", exclude_spans=["receive", "send"], ) meter = metrics.get_meter(__name__) transactions_total = meter.create_counter( "ledgerline.transactions.total", unit="{transaction}", description="Transactions submitted, by final outcome", ) transaction_duration = meter.create_histogram( "ledgerline.transaction.duration", unit="s", description="End-to-end time to process a submitted transaction", # Seconds, not the SDK's millisecond-shaped defaults (0, 5, 10, ... 10000). explicit_bucket_boundaries_advisory=[ 0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10 ], ) # currency is unvalidated client input; only the seeded # set becomes a label value, so a bad client can't mint new series. KNOWN_CURRENCIES = {"EUR", "USD", "GBP"} STATIC_DIR = Path(__file__).parent / "static" app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") # A shared, connection-pooling client for the simple read-only proxy routes # below -- opening a fresh httpx.AsyncClient (and TCP connection) per request # is real, needless overhead under concurrent load. _proxy_client = httpx.AsyncClient( timeout=settings.downstream_timeout_seconds, limits=httpx.Limits(max_connections=100, max_keepalive_connections=50), ) DOWNSTREAM_URLS = { "ledger": settings.ledger_url, "fraud": settings.fraud_url, "sanctions": settings.sanctions_url, "insights": settings.insights_url, "notifier": settings.notifier_url, } class TransactionRequest(BaseModel): account_id: str kind: str amount_minor: int currency: str merchant_name: str category_id: int | None = None @field_validator("kind") @classmethod def kind_must_be_known(cls, v): if v not in ("deposit", "withdrawal", "purchase"): raise ValueError("kind must be one of deposit, withdrawal, purchase") return v @field_validator("amount_minor") @classmethod def deposit_amount_must_be_non_negative(cls, v, info): if info.data.get("kind") == "deposit" and v < 0: raise ValueError("amount_minor must be non-negative for a deposit") return v class ChaosFlagRequest(BaseModel): flag: str enabled: bool class ChaosProfileRequest(BaseModel): profile: str @app.get("/", include_in_schema=False) async def index(): """Serves the frontend (phase-0-base-project-spec.md §10). This overrides §4's generic `GET / -> {"service", "version"}` info route, which the other five HTTP services still expose as-is.""" return FileResponse(STATIC_DIR / "index.html") @app.get("/healthz") async def healthz(): return {"status": "ok", "service": SERVICE_NAME} @app.get("/readyz") async def readyz(): return {"status": "ok"} @app.get("/api/accounts") async def list_accounts(): resp = await _proxy_client.get(f"{settings.ledger_url}/internal/accounts") return JSONResponse(status_code=resp.status_code, content=resp.json()) @app.get("/api/accounts/{account_id}/insights") async def account_insights(account_id: str): resp = await _proxy_client.get(f"{settings.insights_url}/internal/accounts/{account_id}/summary") return JSONResponse(status_code=resp.status_code, content=resp.json()) @app.post("/api/transactions", status_code=201) async def submit_transaction(body: TransactionRequest): request_id = _request_id() start = time.perf_counter() status = "failed" try: result = await orchestrator.create_transaction(body.model_dump()) status = result["status"] except UpstreamFailure as exc: log = logger.error if exc.status_code >= 500 else logger.warning log( "transaction_failed", account_id=body.account_id, status_code=exc.status_code, reason=exc.message, cause=repr(exc.__cause__) if exc.__cause__ else None, ) return JSONResponse( status_code=exc.status_code, content={"error": exc.message, "request_id": request_id} ) finally: transaction_duration.record(time.perf_counter() - start, {"kind": body.kind}) currency = body.currency if body.currency in KNOWN_CURRENCIES else "other" # Phase 5 step 4: deliberate cardinality explosion. Revert after the experiment. transactions_total.add( 1, {"kind": body.kind, "status": status, "currency": currency, "account_id": body.account_id}, ) return JSONResponse(status_code=201, content={**result, "request_id": request_id}) def _request_id() -> str: """The response's `request_id` is the trace ID: paste it into Tempo to get the whole request. 32 hex digits, which is still a valid UUID. Falls back to a random one if no span is recording (telemetry not set up).""" ctx = trace.get_current_span().get_span_context() return format(ctx.trace_id, "032x") if ctx.is_valid else uuid.uuid4().hex @app.get("/api/health/all") async def health_all(): async def check(name, url): try: resp = await _proxy_client.get(f"{url}/healthz", timeout=2.0) return name, "ok" if resp.status_code == 200 else "down" except httpx.RequestError: return name, "down" # The route itself is in excluded_urls; without this its five downstream # /healthz calls would each still be a parentless root trace. with suppress_instrumentation(): results = await asyncio.gather(*[check(name, url) for name, url in DOWNSTREAM_URLS.items()]) services = {"gateway": "ok", **dict(results)} return {"services": services} @app.get("/api/admin/chaos") async def get_chaos(): return {"flags": await get_flags()} @app.post("/api/admin/chaos") async def post_chaos(body: ChaosFlagRequest): if body.flag not in ALL_FLAGS: return JSONResponse(status_code=400, content={"error": f"unknown flag {body.flag!r}"}) await set_flag(body.flag, body.enabled) return {"flags": await get_flags()} @app.post("/api/admin/chaos/profile") async def post_chaos_profile(body: ChaosProfileRequest): try: flags = await set_profile(body.profile) except ValueError as exc: return JSONResponse(status_code=400, content={"error": str(exc)}) return {"flags": flags}