El gateway: FastAPI en producción

Qué vas a provocar: Calcular tú mismo los números de bolsillo: las frases que manda uvicorn, Little, la RAM del peor caso, la cadena de timeouts y la amplificación de reintentos.

Necesitas: solo Python 3.13 (la versión del C01); no instala nada

1. Paso 1Copiar

   mkdir -p ~/lab-gateway && cd ~/lab-gateway

   Qué deberías ver: Sin salida: es la carpeta de los labs de esta hoja.

2. Paso 2Copiar

   python3.13 -c '
   from http import HTTPStatus
   for c in (200, 400, 413, 415, 422, 429, 500, 502, 503, 504):
       print(c, HTTPStatus(c).phrase)
   '

   Qué deberías ver: Diez líneas, de 200 OK a 504 Gateway Timeout. Fíjate en 413 Content Too Large y 422 Unprocessable Content: son los nombres del RFC 9110, que Python usa desde 3.13. Con 3.12 o anterior verías Request Entity Too Large y Unprocessable Entity, y uvicorn mandaría esas frases.

3. Paso 3Copiar

   cat > numeros.py <<'EOF'
   # Números de bolsillo del gateway del C01 (todo en ms, MB y req/s)
   lam = 500 / 5                                  # pico de diseño repartido en 5 pods
   for w in (0.05, 0.2):                          # latencia media: normal y degradada
       print(f"Little: {lam:.0f} req/s × {w} s = {lam * w:.0f} en vuelo por pod")
   print(f"Llamadas abiertas a Triton si se frena: {lam:.0f} × 0.15 s = {lam * 0.15:.0f} por pod")
   ram = 80 + 2 * 320 + 16 * 10                   # Python + 2 decodes de 40 MP + 16 bodies de 10 MB
   print(f"RAM del peor caso: 80 + 2 × 320 + 16 × 10 = {ram} MB de {2**30 / 1e6:.0f} MB (1 GiB)")
   cadena = [300, 250, 150, 100]                  # caller > deadline > Triton > cola de Triton
   print("Timeouts:", " > ".join(f"{t}" for t in cadena), "ms:", all(a > b for a, b in zip(cadena, cadena[1:])))
   for gastado in (40, 130, 260):                 # ms ya usados antes de llamar a Triton
       queda = 250 - gastado
       print(f"gastado {gastado} → queda {queda} → a Triton:", f"{min(150, queda)} ms" if queda > 0 else "504 sin llamar")
   for capas in (1, 2, 3):
       print(f"3 reintentos en {capas} capa(s): hasta {4 ** capas} intentos por request")
   EOF
   python3.13 numeros.py

   Qué deberías ver: Little: 100 req/s × 0.05 s = 5 en vuelo por pod · Little: 100 req/s × 0.2 s = 20 en vuelo por pod · Llamadas abiertas a Triton si se frena: 100 × 0.15 s = 15 por pod · RAM del peor caso: 80 + 2 × 320 + 16 × 10 = 880 MB de 1074 MB (1 GiB) · Timeouts: 300 > 250 > 150 > 100 ms: True · gastado 40 → queda 210 → a Triton: 150 ms · gastado 130 → queda 120 → a Triton: 120 ms · gastado 260 → queda -10 → a Triton: 504 sin llamar · y 4, 16 y 64 intentos con 1, 2 y 3 capas que reintentan. Cada línea es una fila de esta sección: Little explica el 16, la RAM explica los 2 decodes, y el 64 explica por qué reintenta una sola capa.

4. Paso 4Copiar

   cd ~ && rm -rf ~/lab-gateway

   Qué deberías ver: Sin salida.

Probado de verdad en macOS (Apple M4), con Python 3.13.16 y 3.14.3. La misma salida. Las frases de HTTPStatus se compararon también con Python 3.9.6, que da Request Entity Too Large y Unprocessable Entity.

Qué vas a provocar: Armar una cadena en miniatura (admisión, validación, CPU en el threadpool, un Triton falso y el log en el finally) y comprobar qué requests llegan a tu log y cuáles no.

Necesitas: Python 3.13 (la versión del C01) e internet para pip: FastAPI, Starlette, Pydantic y uvicorn, en las versiones del C01. curl ya viene en macOS. Sin Docker ni GPU.

1. Paso 1Copiar

   mkdir -p ~/lab-gateway && cd ~/lab-gateway
   python3.13 -m venv .venv && . .venv/bin/activate
   pip install fastapi==0.141.1 starlette==1.7.0 pydantic==2.13.5 uvicorn==0.54.0

   Qué deberías ver: La última línea empieza con Successfully installed y trae fastapi-0.141.1, starlette-1.7.0, pydantic-2.13.5 y uvicorn-0.54.0, más sus dependencias (pydantic-core, anyio, h11, click…). Son las versiones del C01. Si ya tenías el venv de otro lab, dice Requirement already satisfied.

2. Paso 2Copiar

   cat > cadena.py <<'EOF'
   import asyncio, json, logging, sys, time, uuid
   from contextlib import asynccontextmanager
   
   from fastapi import FastAPI, HTTPException, Query, Request, Response
   from starlette.concurrency import run_in_threadpool
   
   MAX_INFLIGHT = 2                               # 16 en el C01; 2 para verlo con 6 requests
   log = logging.getLogger("lab")                 # como el C01: un logger propio a stdout
   log.addHandler(logging.StreamHandler(sys.stdout))
   log.setLevel(logging.INFO)
   log.propagate = False
   
   @asynccontextmanager
   async def lifespan(app: FastAPI):
       app.state.inflight = 0
       yield
   
   app = FastAPI(lifespan=lifespan)
   
   def decode(data: bytes) -> int:                # CPU simulada: corre en el threadpool
       time.sleep(0.05)
       return len(data)
   
   async def triton() -> list[float]:             # I/O simulado: el await cede el turno
       await asyncio.sleep(0.02)
       return [0.7, 0.2, 0.1]
   
   @app.post("/v1/classify")
   async def classify(request: Request, response: Response, top_k: int = Query(5, ge=1, le=20)):
       t0 = time.perf_counter()
       rid = request.headers.get("x-request-id") or uuid.uuid4().hex
       status = 500
       try:
           if request.app.state.inflight >= MAX_INFLIGHT:
               raise HTTPException(503, "gateway saturado", headers={"Retry-After": "1"})
           request.app.state.inflight += 1
           try:
               if request.headers.get("content-type") != "image/jpeg":
                   raise HTTPException(415, "Content-Type debe ser image/jpeg")
               data = await request.body()        # el lab no pone tope; el C01 cuenta los bytes
               n = await run_in_threadpool(decode, data)
               scores = await triton()
           finally:
               request.app.state.inflight -= 1
           status = 200
           response.headers["X-Request-ID"] = rid
           return {"bytes": n, "scores": scores[:top_k]}
       except HTTPException as exc:
           status = exc.status_code
           raise
       finally:
           ms = round((time.perf_counter() - t0) * 1000, 1)
           log.info(json.dumps({"request_id": rid, "status": status, "ms": ms}))
   EOF

   Qué deberías ver: Sin salida: el archivo queda en ~/lab-gateway/cadena.py. Admite 2 requests en vuelo, simula 50 ms de decode en un hilo y 20 ms de Triton, y escribe una línea JSON por request en el finally.

3. Paso 3Copiar

   uvicorn cadena:app --port 8080 --no-access-log > server.log 2>&1 &
   echo $! > uvicorn.pid
   sleep 2; cat server.log

   Qué deberías ver: Cuatro líneas de uvicorn: Started server process [PID], Waiting for application startup., Application startup complete. y Uvicorn running on http://127.0.0.1:8080 (Press CTRL+C to quit). La terminal muestra además el número del job en segundo plano.

4. Paso 4Copiar

   curl -si -H "Content-Type: image/jpeg" --data-binary "hola" "localhost:8080/v1/classify?top_k=2"
   echo
   curl -s -w ' %{http_code}\n' -H "Content-Type: image/png" --data-binary "hola" localhost:8080/v1/classify
   curl -s -w ' %{http_code}\n' -H "Content-Type: image/jpeg" --data-binary "hola" "localhost:8080/v1/classify?top_k=0"

   Qué deberías ver: Primero HTTP/1.1 200 OK, los headers (entre ellos x-request-id con 32 caracteres hexadecimales: Starlette, la base de FastAPI, guarda los nombres en minúsculas) y el body {"bytes":4,"scores":[0.7,0.2]}. Después {"detail":"Content-Type debe ser image/jpeg"} 415. Y por último {"detail":[{"type":"greater_than_equal","loc":["query","top_k"],"msg":"Input should be greater than or equal to 1","input":"0","ctx":{"ge":1}}]} 422.

5. Paso 5Copiar

   seq 6 | xargs -P 6 -I{} curl -s -o /dev/null -w '%{http_code}\n' -H "Content-Type: image/jpeg" --data-binary x localhost:8080/v1/classify | sort | uniq -c

   Qué deberías ver: Seis requests a la vez. Por lo general salen 2 × 200 y 4 × 503: solo caben 2 en vuelo y cada uno tarda ≈ 70 ms. Si tu Mac lanza los curl más despacio, puede colarse un 200 más.

6. Paso 6Copiar

   grep -o '"status": [0-9]*' server.log | sort | uniq -c

   Qué deberías ver: Los status que vio el finally: los 200 (el primero y los de la ráfaga), un 415 y los 503. Ningún 422: el request con top_k=0 nunca entró a classify, así que su finally no corrió y no dejó línea. El C01 lo resuelve con un exception handler para el 422.

7. Paso 7Copiar

   kill $(cat uvicorn.pid); sleep 1; tail -4 server.log

   Qué deberías ver: INFO: Shutting down, Waiting for application shutdown., Application shutdown complete. y Finished server process [PID]: uvicorn atrapó el SIGTERM de kill y se apagó en orden.

8. Paso 8Copiar

   deactivate; cd ~ && rm -rf ~/lab-gateway

   Qué deberías ver: Sin salida: borra el venv, el código y server.log.

Verificado contra la documentación; no ejecutado aquí. Sin instalar nada: cadena.py se corrió en macOS con FastAPI 0.143.0, Starlette 1.7.0, Pydantic 2.14.1 y uvicorn 0.54.0 (con h11 y asyncio, como en el venv del lab) y dio exactamente estas salidas. Para FastAPI 0.141.1 y Pydantic 2.13.5, las versiones del C01, se verificó en el código: el 422 sale antes de llamar a la función, con {"detail": errores} y sin la clave url.

Qué vas a provocar: Encontrar en el código real del C01 cada responsabilidad del gateway (errores, configuración, rutas publicadas) y calcular cuánta CPU y red cuesta separarlo de Triton.

Necesitas: el repo del curso y Python 3.13; no instala nada ni crea archivos

1. Paso 1Copiar

   cd apps/ml-system-design/reference/c01
   grep -oE 'HTTPException\([0-9]{3}' gateway/app/main.py | sort | uniq -c

   Qué deberías ver: 2 HTTPException(400 · 3 HTTPException(413 · 1 HTTPException(415 · 1 HTTPException(499 · 1 HTTPException(500 · 5 HTTPException(503 · 2 HTTPException(504: los errores que el gateway decide. El 499 es el cliente que corta a mitad del upload, y el 500, Triton devolviendo logits con NaN. No hay 422: lo genera FastAPI (el handler del C01 solo lo anota). Un bug sin atrapar también sale como 500, pero lo responde Starlette, no esta lista.

2. Paso 2Copiar

   grep -c '"Retry-After": "1"' gateway/app/main.py

   Qué deberías ver: 4: cuatro de los cinco 503 traen Retry-After. El quinto es el de /readyz, que lo lee el kubelet, no un cliente.

3. Paso 3Copiar

   grep -oE 'getenv\("[A-Z_]+"' gateway/app/main.py

   Qué deberías ver: Ocho líneas: TRITON_URL, MODEL_NAME, MODEL_VERSION, INFER_TIMEOUT_S, LABELS_PATH, PREPROCESS_CONCURRENCY, MAX_INFLIGHT y REQUEST_DEADLINE_S, las perillas que se cambian sin reconstruir la imagen.

4. Paso 4Copiar

   grep -nE '^@app\.' gateway/app/main.py
   grep -nE 'path: /v1|pathType' k8s/gateway.yaml

   Qué deberías ver: 119 @app.get("/metrics", include_in_schema=False), 206 @app.post("/v1/classify"), 319 @app.exception_handler(RequestValidationError), 342 @app.get("/healthz") y 348 @app.get("/readyz"); y en el Ingress, 186 path: /v1/classify con 187 pathType: Exact. Cuatro rutas en el pod (la 319 no es una ruta: es el handler del 422), una sola publicada.

5. Paso 5Copiar

   python3.13 -c '
   lam, t_cpu = 500, 0.006                 # req/s en el pico de diseño; CPU por foto (≈ 6 ms)
   cores = lam * t_cpu
   print(f"CPU de decode: {lam} × {t_cpu} s = {cores:.0f} cores")
   print(f"Pods del gateway al 60 % de 1 core: {cores / 0.6:.0f}")
   print(f"Red gateway → Triton: {lam} × 602,112 B × 8 = {lam * 602112 * 8 / 1e9:.1f} Gb/s")
   '

   Qué deberías ver: CPU de decode: 500 × 0.006 s = 3 cores · Pods del gateway al 60 % de 1 core: 5 · Red gateway → Triton: 500 × 602,112 B × 8 = 2.4 Gb/s. Los 3 cores, en el nodo con GPU, competirían con Triton por 4 vCPU; separados, los pone el HPA en nodos de CPU.

Probado de verdad en un Mac. Ejecutado en macOS con el grep de BSD y Python 3.13.16 sobre los archivos del repo, en octubre de 2026.

Qué vas a provocar: Escribir un servicio con lifespan, un modelo de salida, una dependencia y una ruta fuera del schema, y ver qué valida, qué filtra y qué publica FastAPI en OpenAPI.

Necesitas: Python 3.13 (la versión del C01) e internet para pip: FastAPI, Starlette, Pydantic y uvicorn, en las versiones del C01. curl ya viene en macOS. Sin Docker ni GPU.

1. Paso 1Copiar

   mkdir -p ~/lab-gateway && cd ~/lab-gateway
   python3.13 -m venv .venv && . .venv/bin/activate
   pip install fastapi==0.141.1 starlette==1.7.0 pydantic==2.13.5 uvicorn==0.54.0

   Qué deberías ver: La última línea empieza con Successfully installed y trae fastapi-0.141.1, starlette-1.7.0, pydantic-2.13.5 y uvicorn-0.54.0, más sus dependencias (pydantic-core, anyio, h11, click…). Son las versiones del C01. Si ya tenías el venv de otro lab, dice Requirement already satisfied.

2. Paso 2Copiar

   cat > basico.py <<'EOF'
   import logging
   from contextlib import asynccontextmanager
   from typing import Annotated
   
   from fastapi import Depends, FastAPI, HTTPException, Query, Request
   from pydantic import BaseModel
   
   log = logging.getLogger("uvicorn.error")       # sale con el mismo formato que los logs de uvicorn
   
   class Prediccion(BaseModel):
       label: str
       score: float
   
   class Salida(BaseModel):
       model_version: str
       predictions: list[Prediccion]
   
   @asynccontextmanager
   async def lifespan(app: FastAPI):
       log.info("lifespan: cargo los labels")
       app.state.labels = ["Smartphones", "Fundas para celular", "Cargadores de celular"]
       yield
       log.info("lifespan: cierro todo")
   
   app = FastAPI(title="lab-gateway", lifespan=lifespan)
   
   def get_labels(request: Request) -> list[str]:
       return request.app.state.labels
   
   Labels = Annotated[list[str], Depends(get_labels)]
   
   @app.post("/v1/classify")
   async def classify(labels: Labels, top_k: Annotated[int, Query(ge=1, le=3)] = 2) -> Salida:
       if not labels:
           raise HTTPException(503, "sin labels", headers={"Retry-After": "1"})
       preds = [{"label": l, "score": s, "debug": "interno"} for l, s in zip(labels, (0.83, 0.07, 0.04))]
       return {"model_version": "lab-1", "predictions": preds[:top_k], "costo_ms": 3}
   
   @app.get("/healthz", include_in_schema=False)
   async def healthz():
       return {"status": "ok"}
   EOF

   Qué deberías ver: Sin salida: el archivo queda en ~/lab-gateway/basico.py. La ruta devuelve dos campos que el modelo Salida no declara (costo_ms, y debug en cada predicción).

3. Paso 3Copiar

   uvicorn basico:app --port 8080 > server.log 2>&1 &
   echo $! > uvicorn.pid
   sleep 2; cat server.log

   Qué deberías ver: Las líneas de arranque de uvicorn y, entre Waiting for application startup. y Application startup complete., la tuya: INFO: lifespan: cargo los labels. El lifespan corre antes de aceptar requests.

4. Paso 4Copiar

   curl -s -X POST "localhost:8080/v1/classify?top_k=2"
   echo
   curl -s -w ' %{http_code}\n' -X POST "localhost:8080/v1/classify?top_k=9"

   Qué deberías ver: {"model_version":"lab-1","predictions":[{"label":"Smartphones","score":0.83},{"label":"Fundas para celular","score":0.07}]}: sin costo_ms ni debug, porque el modelo de salida los filtró. Después {"detail":[{"type":"less_than_equal","loc":["query","top_k"],"msg":"Input should be less than or equal to 3","input":"9","ctx":{"le":3}}]} 422.

5. Paso 5Copiar

   curl -s localhost:8080/openapi.json | python3 -c '
   import json, sys
   d = json.load(sys.stdin)
   print(sorted(d["paths"]))
   p = d["paths"]["/v1/classify"]["post"]["parameters"][0]
   print(p["name"], p["in"], p["schema"]["minimum"], p["schema"]["maximum"], p["schema"]["default"])
   print(sorted(d["components"]["schemas"]))
   '

   Qué deberías ver: ['/v1/classify'] (sin /healthz, por include_in_schema=False), después top_k query 1 3 2, y ['HTTPValidationError', 'Prediccion', 'Salida', 'ValidationError']: el rango de top_k, los modelos de la respuesta y el formato del 422 quedan documentados sin escribir nada. La request (Request) y la dependencia no aparecen como parámetros. Abre http://localhost:8080/docs en el navegador para verlo como página.

6. Paso 6Copiar

   kill $(cat uvicorn.pid); sleep 1; tail -5 server.log

   Qué deberías ver: INFO: Shutting down, Waiting for application shutdown., INFO: lifespan: cierro todo, Application shutdown complete. y Finished server process [PID]: lo de después del yield corre al apagar.

7. Paso 7Copiar

   deactivate; cd ~ && rm -rf ~/lab-gateway

   Qué deberías ver: Sin salida.

Verificado contra la documentación; no ejecutado aquí. Sin instalar nada: basico.py se corrió en macOS con FastAPI 0.143.0, Starlette 1.7.0, Pydantic 2.14.1 y uvicorn 0.54.0 y dio exactamente estas salidas. Para la versión del C01 se verificó en la documentación y el código de FastAPI 0.141.1 (el modelo de salida filtra campos; ge y le salen como minimum y maximum en OpenAPI; include_in_schema=False saca la ruta; el 422 sale antes de la función) y en uvicorn 0.54.0 (orden de los mensajes de arranque y apagado).

Qué vas a provocar: Ver el event loop bloqueado: tres fotos y un /healthz con el decode dentro del loop y en un hilo, el aviso del modo debug y el efecto del GIL en hilos.

Necesitas: solo Python 3.13 (la versión del C01); no instala nada. Con 3.11 o 3.12 también corre, salvo os.process_cpu_count, que es de 3.13.

1. Paso 1Copiar

   mkdir -p ~/lab-gateway && cd ~/lab-gateway

   Qué deberías ver: Sin salida.

2. Paso 2Copiar

   cat > bloqueo.py <<'EOF'
   import asyncio, time
   
   DECODE_S, TRITON_S = 0.05, 0.01        # 50 ms de CPU por foto; 10 ms esperando a Triton
   
   def decode():                           # simula el decode: no tiene await (Pillow, igual, suelta el GIL)
       time.sleep(DECODE_S)
   
   async def foto_mal(t0, nombre, fin, slots):
       decode()                            # MAL: corre en el event loop y no suelta el turno
       await asyncio.sleep(TRITON_S)       # I/O: aquí sí cede el turno
       fin[nombre] = round((time.perf_counter() - t0) * 1000)
   
   async def foto_bien(t0, nombre, fin, slots):
       async with slots:                   # como el C01: 2 decodes a la vez
           await asyncio.to_thread(decode) # BIEN: un hilo decodifica; el loop sigue libre
       await asyncio.sleep(TRITON_S)
       fin[nombre] = round((time.perf_counter() - t0) * 1000)
   
   async def escenario(foto):
       asyncio.get_running_loop().slow_callback_duration = 0.02  # modo debug: avisa desde 20 ms (default: 100 ms)
       t0, fin, slots = time.perf_counter(), {}, asyncio.Semaphore(2)
       llega = t0 + 0.005                  # un /healthz llega a los 5 ms
       def responder_healthz():
           fin["healthz"] = round((time.perf_counter() - llega) * 1000)
       asyncio.get_running_loop().call_later(0.005, responder_healthz)
       await asyncio.gather(*(foto(t0, f"foto{i}", fin, slots) for i in (1, 2, 3)))
       return fin
   
   for titulo, foto in (("mal ", foto_mal), ("bien", foto_bien)):
       fin = asyncio.run(escenario(foto))
       print(titulo, "ms hasta responder:", dict(sorted(fin.items(), key=lambda kv: kv[1])))
   EOF
   python3.13 bloqueo.py

   Qué deberías ver: Dos líneas. mal: healthz ≈ 155–180 ms, foto1 y foto2 ≈ 160–185, foto3 ≈ 170–195: cada decode de 50 ms bloquea a todos, y el /healthz que llegó a los 5 ms espera los tres. bien: healthz 0 o 1, foto1 y foto2 ≈ 60–75 (decodifican juntas, en dos hilos) y foto3 ≈ 115–135 (esperó un slot). No da 145 · 150 · 160 exactos porque time.sleep en macOS se pasa 5–10 ms en cada llamada, más si la Mac está ocupada.

3. Paso 3Copiar

   PYTHONASYNCIODEBUG=1 python3.13 bloqueo.py 2>&1 | grep -oE 'coro=<foto_[a-z]+\(\)|took [0-9.]+ seconds' | paste - -

   Qué deberías ver: Tres líneas, coro=<foto_mal() con took ≈ 0.055–0.060 seconds: el modo debug dice qué corrutina tuvo el loop ocupado. En el escenario bien no hay ninguna. El script baja el umbral a 20 ms; con el default de 100 ms, estos bloqueos de 50 ms no aparecerían.

4. Paso 4Copiar

   cat > gil.py <<'EOF'
   import hashlib, os, threading, time
   
   datos = os.urandom(32 * 1024 * 1024)        # 32 MB al azar
   
   def hashear():                               # C que suelta el GIL (como el sha256 del C01)
       for _ in range(8):
           hashlib.sha256(datos).hexdigest()
   
   def sumar():                                 # Python puro: necesita el GIL en cada paso
       n = 0
       for i in range(3_000_000):
           n += i
   
   def medir(f, hilos):
       t = time.perf_counter()
       ts = [threading.Thread(target=f) for _ in range(hilos)]
       for x in ts: x.start()
       for x in ts: x.join()
       return time.perf_counter() - t
   
   for f in (hashear, sumar):
       uno, cuatro = medir(f, 1), medir(f, 4)
       print(f"{f.__name__}: 1 hilo {uno:.2f} s · 4 hilos {cuatro:.2f} s → {4 * uno / cuatro:.1f}× trabajo por segundo")
   print("CPUs:", os.process_cpu_count(), "· pool de asyncio.to_thread:", min(32, (os.process_cpu_count() or 1) + 4), "hilos")
   EOF
   python3.13 gil.py

   Qué deberías ver: hashear: 4 hilos tardan lo mismo que 1 (≈ 0.17 s en una M4) → ≈ 4× trabajo por segundo (hashlib suelta el GIL, como Pillow en el decode). sumar: 1 hilo ≈ 0.09 s · 4 hilos ≈ 0.36 s → 1.0× (Python puro: un hilo a la vez). La última línea dice tus núcleos y el tamaño del pool de asyncio.to_thread: en una Mac M4, 10 y 14 hilos.

5. Paso 5Copiar

   cd ~ && rm -rf ~/lab-gateway

   Qué deberías ver: Sin salida.

Probado de verdad en un Mac. Ejecutado varias veces con Python 3.13.16 y 3.14.3 en macOS (Apple M4, 10 núcleos): los tiempos varían entre 5 y 15 ms de una corrida a otra, según la carga de la Mac (con la Mac ocupada, time.sleep(0.05) llegó a durar ≈ 59 ms). El texto del aviso del modo debug es el de asyncio en 3.13 y 3.14.

Qué vas a provocar: Provocar cada código del contrato con curl: 200, 415, 413 por Content-Length y contando bytes, 400, y los 422 de Pydantic (rango, campo de más, JSON roto y Content-Type olvidado).

Necesitas: Python 3.13 (la versión del C01) e internet para pip: FastAPI, Starlette, Pydantic y uvicorn, en las versiones del C01. curl ya viene en macOS. Sin Docker ni GPU.

1. Paso 1Copiar

   mkdir -p ~/lab-gateway && cd ~/lab-gateway
   python3.13 -m venv .venv && . .venv/bin/activate
   pip install fastapi==0.141.1 starlette==1.7.0 pydantic==2.13.5 uvicorn==0.54.0

   Qué deberías ver: La última línea empieza con Successfully installed y trae fastapi-0.141.1, starlette-1.7.0, pydantic-2.13.5 y uvicorn-0.54.0, más sus dependencias (pydantic-core, anyio, h11, click…). Son las versiones del C01. Si ya tenías el venv de otro lab, dice Requirement already satisfied.

2. Paso 2Copiar

   cat > contrato.py <<'EOF'
   from typing import Annotated
   
   from fastapi import FastAPI, HTTPException, Request
   from pydantic import BaseModel, ConfigDict, Field
   
   MAX_BYTES = 1000                               # 1 KB en el lab; 10 MB en el C01
   TIPOS = {"image/jpeg", "image/png"}
   app = FastAPI()
   
   class Opciones(BaseModel):
       model_config = ConfigDict(extra="forbid")
       top_k: Annotated[int, Field(ge=1, le=20)] = 5
   
   @app.post("/v1/opciones")
   async def opciones(o: Opciones) -> Opciones:
       return o
   
   @app.post("/v1/subir")
   async def subir(request: Request):
       ctype = request.headers.get("content-type", "").split(";")[0].strip().lower()
       if ctype not in TIPOS:
           raise HTTPException(415, "Content-Type debe ser image/jpeg o image/png")
       declarado = request.headers.get("content-length", "")
       if declarado.isdigit() and int(declarado) > MAX_BYTES:
           raise HTTPException(413, "supera 1 KB (por Content-Length)")
       total = 0
       async for chunk in request.stream():
           total += len(chunk)
           if total > MAX_BYTES:
               raise HTTPException(413, "supera 1 KB (contando bytes)")
       if total == 0:
           raise HTTPException(400, "body vacío")
       return {"bytes": total}
   EOF

   Qué deberías ver: Sin salida: el archivo queda en ~/lab-gateway/contrato.py, con un endpoint de bytes crudos al estilo del C01 (tope de 1 KB) y uno JSON con un modelo de Pydantic.

3. Paso 3Copiar

   uvicorn contrato:app --port 8080 --no-access-log > server.log 2>&1 &
   echo $! > uvicorn.pid
   sleep 2; cat server.log
   head -c 500 /dev/zero > chica.bin && head -c 2000 /dev/zero > grande.bin

   Qué deberías ver: Las cuatro líneas de arranque de uvicorn. Los dos archivos quedan creados: 500 y 2,000 bytes en cero.

4. Paso 4Copiar

   U=localhost:8080/v1/subir
   curl -s -w ' %{http_code}\n' -H "Content-Type: image/jpeg" --data-binary @chica.bin $U
   curl -s -w ' %{http_code}\n' -H "Content-Type: image/gif" --data-binary @chica.bin $U
   curl -s -w ' %{http_code}\n' -H "Content-Type: image/jpeg" --data-binary @grande.bin $U
   curl -s -w ' %{http_code}\n' -H "Content-Type: image/jpeg" -H "Transfer-Encoding: chunked" --data-binary @grande.bin $U
   curl -s -w ' %{http_code}\n' -H "Content-Type: image/jpeg" --data-binary "" $U

   Qué deberías ver: {"bytes":500} 200 · {"detail":"Content-Type debe ser image/jpeg o image/png"} 415 · {"detail":"supera 1 KB (por Content-Length)"} 413 · {"detail":"supera 1 KB (contando bytes)"} 413 · {"detail":"body vacío"} 400. El cuarto llega con chunked, sin Content-Length: el primer control lo deja pasar y lo frena el segundo, el que cuenta los bytes.

5. Paso 5Copiar

   U=localhost:8080/v1/opciones; J='Content-Type: application/json'
   curl -s -w ' %{http_code}\n' -H "$J" -d '{"top_k": 50}' $U
   curl -s -w ' %{http_code}\n' -H "$J" -d '{"top_k": 3, "debug": true}' $U
   curl -s -w ' %{http_code}\n' -H "$J" -d '{"top_k": "3"}' $U
   curl -s -w ' %{http_code}\n' -H "$J" -d '{"top_k": 3' $U
   curl -s -o /dev/null -w '%{http_code}\n' -d '{"top_k": 3}' $U

   Qué deberías ver: {"detail":[{"type":"less_than_equal","loc":["body","top_k"],"msg":"Input should be less than or equal to 20","input":50,"ctx":{"le":20}}]} 422 · {"detail":[{"type":"extra_forbidden","loc":["body","debug"],"msg":"Extra inputs are not permitted","input":true}]} 422 · {"top_k":3} 200 (modo lax: el texto "3" se convirtió en 3) · {"detail":[{"type":"json_invalid","loc":["body",11],"msg":"JSON decode error","input":{},"ctx":{"error":"Expecting ',' delimiter"}}]} 422, un JSON roto también es 422 · y 422 a secas: sin -H, curl -d manda Content-Type: application/x-www-form-urlencoded, y FastAPI no lee como JSON un body con ese Content-Type (si el header falta del todo, desde 0.132 también es 422).

6. Paso 6Copiar

   kill $(cat uvicorn.pid); sleep 1; tail -4 server.log

   Qué deberías ver: Las cuatro líneas del apagado de uvicorn, de Shutting down a Finished server process [PID].

7. Paso 7Copiar

   deactivate; cd ~ && rm -rf ~/lab-gateway

   Qué deberías ver: Sin salida.

Verificado contra la documentación; no ejecutado aquí. Sin instalar nada: contrato.py se corrió en macOS con FastAPI 0.143.0, Starlette 1.7.0, Pydantic 2.14.1 y uvicorn 0.54.0 y dio exactamente estas salidas, incluidos el 413 por chunked y el json_invalid en la posición 11. Para la versión del C01 se verificó en el código de FastAPI 0.141.1 (JSON roto → json_invalid con loc ["body", posición] y msg JSON decode error; sin Content-Type de JSON, el body no se lee como JSON; 422 con {"detail": errores} sin url), en los mensajes de pydantic-core (less than or equal to, Extra inputs are not permitted) y en Python 3.13, que da Expecting ',' delimiter en la posición 11 para ese JSON.

Qué vas a provocar: Ver, con un Triton falso hecho solo con Python estándar, cómo el deadline recorta el tiempo de cada salto, cuántas llamadas quedan abiertas con y sin timeout (Little) y qué hace el jitter con una tormenta de reintentos.

Necesitas: Python 3.11 o más nuevo (los comandos usan python3.13, la versión de C01), sin paquetes extra. Ni Docker ni GPU.

1. Paso 1Copiar

   mkdir -p ~/lab-gateway && cd ~/lab-gateway
   cat > deadline.py <<'EOF'
   # deadline.py: un "Triton" falso y un gateway con deadline, como en C01 (solo stdlib)
   import asyncio, time
   
   INFER_TIMEOUT_S = 0.15  # tope del salto a Triton (C01)
   DEADLINE_S = 0.25       # presupuesto total del request (C01)
   
   async def triton_falso(demora_s):
       await asyncio.sleep(demora_s)  # cola + GPU, simuladas
       return "logits"
   
   async def gateway(decode_s, triton_s):
       t0 = time.perf_counter()
       deadline = t0 + DEADLINE_S                  # se fija una vez, al entrar
       await asyncio.sleep(decode_s)               # el decode (en C01 va al threadpool)
       left = deadline - time.perf_counter()       # lo que queda del presupuesto
       if left <= 0:
           return 504, "se agotó el presupuesto", t0
       try:
           await asyncio.wait_for(triton_falso(triton_s), timeout=min(INFER_TIMEOUT_S, left))
           return 200, f"ok (Triton tenía {min(INFER_TIMEOUT_S, left) * 1000:.0f} ms)", t0
       except TimeoutError:
           return 504, f"timeout de inferencia (tenía {min(INFER_TIMEOUT_S, left) * 1000:.0f} ms)", t0
   
   async def main():
       casos = [("normal", 0.010, 0.008), ("Triton lento", 0.010, 1.0),
                ("decode lento", 0.200, 0.080), ("decode eterno", 0.300, 0.008)]
       for nombre, dec, tri in casos:
           codigo, detalle, t0 = await gateway(dec, tri)
           dt = (time.perf_counter() - t0) * 1000
           print(f"{nombre:13} -> {codigo} {detalle:38} respondió en {dt:4.0f} ms")
   
   asyncio.run(main())
   EOF
   python3.13 deadline.py

   Qué deberías ver: Cuatro líneas. normal → 200 en ≈ 20 ms (Triton tenía 150 ms). Triton lento → 504 timeout de inferencia en ≈ 160 ms: el gateway no espera el segundo entero. decode lento → 504 en ≈ 250 ms, y Triton tenía solo ≈ 49 ms: lo que quedaba del presupuesto. decode eterno → 504 se agotó el presupuesto en ≈ 300 ms: ni se llamó a Triton, pero el decode no se corta, así que el caller (300 ms) ya se fue. Los ms exactos varían 1 o 2 entre corridas.

2. Paso 2Copiar

   cat > envuelo.py <<'EOF'
   # envuelo.py: 100 req/s durante 3 s contra un Triton colgado 1 s; ¿cuántas llamadas abiertas a la vez?
   import asyncio
   
   async def llamada(timeout_s, estado):
       estado["abiertas"] += 1
       estado["max"] = max(estado["max"], estado["abiertas"])
       try:
           await asyncio.wait_for(asyncio.sleep(1.0), timeout=timeout_s)  # Triton no contesta en 1 s
       except TimeoutError:
           pass  # el gateway respondería 504
       finally:
           estado["abiertas"] -= 1
   
   async def carga(timeout_s, tasa=100, segundos=3):
       estado = {"abiertas": 0, "max": 0}
       loop = asyncio.get_running_loop()
       inicio, tareas = loop.time(), []
       for i in range(tasa * segundos):
           await asyncio.sleep(max(0.0, inicio + i / tasa - loop.time()))  # una llamada cada 10 ms
           tareas.append(asyncio.create_task(llamada(timeout_s, estado)))
       await asyncio.gather(*tareas)
       return estado["max"]
   
   async def main():
       for t in (None, 0.15):
           w = 1.0 if t is None else t
           print(f"timeout {str(t):4}: máximo de llamadas abiertas = {await carga(t):3}   (Little: 100 req/s × {w} s = {100 * w:.0f})")
   
   asyncio.run(main())
   EOF
   python3.13 envuelo.py

   Qué deberías ver: Tarda ≈ 7 s. timeout None: máximo de llamadas abiertas = 101 (Little: 100). timeout 0.15: máximo = 16 (Little: 15). El timeout no hace más rápido a Triton: acota cuánto se acumula en el gateway mientras Triton está mal (16 coincide con el MAX_INFLIGHT de C01).

3. Paso 3Copiar

   cat > reintentos.py <<'EOF'
   # reintentos.py: 1,000 clientes fallan juntos en t = 0 y reintentan 3 veces; ¿cuántos golpean a la vez?
   import random
   from collections import Counter
   
   def llegadas(jitter, clientes=1000, reintentos=3, base_ms=100, tope_ms=2000):
       random.seed(7)  # misma semilla = mismos números en cualquier Python 3
       tiempos = []
       for _ in range(clientes):
           t = 0.0
           for n in range(reintentos):
               maximo = min(tope_ms, base_ms * 2 ** n)              # backoff exponencial: 100, 200, 400 ms
               t += random.uniform(0, maximo) if jitter else maximo  # full jitter: al azar entre 0 y el máximo
               tiempos.append(t)
       return tiempos
   
   for jitter in (False, True):
       ventanas = Counter(int(t // 10) for t in llegadas(jitter))  # reintentos por ventana de 10 ms
       nombre = "full jitter" if jitter else "sin jitter "
       print(f"{nombre}: pico = {max(ventanas.values()):4} reintentos en 10 ms; ventanas de 10 ms con tráfico = {len(ventanas)}")
   
   print("amplificación con 3 reintentos por capa y 3 capas:", (1 + 3) ** 3)
   print("amplificación con 2 reintentos por capa y 5 capas:", (1 + 2) ** 5)
   EOF
   python3.13 reintentos.py

   Qué deberías ver: sin jitter: pico = 1000 reintentos en 10 ms y solo 3 ventanas con tráfico (a los 100, 300 y 700 ms): los 1,000 clientes vuelven juntos, tres veces. full jitter: pico = 154, repartidos en 66 ventanas. Y las amplificaciones: 64 con 3 reintentos en 3 capas, 243 con 2 reintentos en 5 capas.

4. Paso 4Copiar

   cd ~ && rm -rf ~/lab-gateway

   Qué deberías ver: Sin salida.

Probado de verdad en un Mac. Ejecutado en macOS (Apple Silicon) con Python 3.13.15, 3.13.16 y 3.14.3, y repetido copiando los pasos tal cual en zsh: las salidas de arriba son las de esas corridas (en deadline.py, ±2 ms entre corridas). reintentos.py da exactamente lo mismo también con Python 3.9.6. Solo usa la biblioteca estándar (asyncio.wait_for, asyncio.sleep, random, collections).

Qué vas a provocar: Comparar, con una simulación determinista hecha solo con Python estándar, un pod sin límite contra uno con las puertas de C01, a tres ritmos de llegada: cuántos salen a tiempo, cuántos reciben 503 y el p99 (el tiempo bajo el que queda el 99 % de los que salen a tiempo).

Necesitas: Python 3, sin paquetes extra (los comandos usan python3.13, la versión de C01). Ni Docker ni GPU.

1. Paso 1Copiar

   mkdir -p ~/lab-gateway && cd ~/lab-gateway
   cat > admision.py <<'EOF'
   # admision.py: un pod del gateway con 2 slots de CPU, con y sin control de admisión (simulación, solo stdlib)
   import heapq, random, sys
   
   SLOTS, MAX_EN_VUELO, ESPERA_SLOT, DEADLINE, TRITON = 2, 16, 0.050, 0.250, 0.008  # números de C01
   
   def simular(tasa, con_admision, segundos=20):
       random.seed(1)  # misma semilla = mismos números en cualquier Python 3
       t, llegadas = 0.0, []
       while t < segundos:                       # llegadas al azar (Poisson): `tasa` requests por segundo
           t += random.expovariate(tasa)
           llegadas.append((t, 0.004 + random.expovariate(1 / 0.008)))  # decode: 4 ms + exponencial de 8 ms
       libres = [0.0] * SLOTS                    # cuándo queda libre cada slot de CPU
       en_vuelo = []                             # cuándo termina cada request aceptado
       a_tiempo, rechazos, tarde, cpu_total, cpu_tirada, lat, pico = 0, 0, 0, 0.0, 0.0, [], 0
       for llega, decode in llegadas:
           while en_vuelo and en_vuelo[0] <= llega:
               heapq.heappop(en_vuelo)
           if con_admision and len(en_vuelo) >= MAX_EN_VUELO:
               rechazos += 1                     # puerta 1: 503 en microsegundos
               continue
           inicio = max(llega, libres[0])        # espera un slot (FIFO)
           if con_admision and inicio - llega > ESPERA_SLOT:
               rechazos += 1                     # puerta 2: sin slot en 50 ms → 503
               heapq.heappush(en_vuelo, llega + ESPERA_SLOT)
               continue
           heapq.heapreplace(libres, inicio + decode)
           fin = inicio + decode + TRITON
           heapq.heappush(en_vuelo, fin)
           pico = max(pico, len(en_vuelo))
           cpu_total += decode
           if fin - llega <= DEADLINE:
               a_tiempo += 1
               lat.append(fin - llega)
           else:
               tarde += 1                        # el caller ya se fue: CPU tirada
               cpu_tirada += decode
       n = len(llegadas)
       lat.sort()
       p99 = lat[int(0.99 * len(lat))] * 1000 if lat else float("nan")
       print(f"{'con admisión' if con_admision else 'sin límite  '}  a tiempo {100 * a_tiempo / n:5.1f} %  "
             f"503 {100 * rechazos / n:5.1f} %  tarde {100 * tarde / n:5.1f} %  "
             f"p99 a tiempo {p99:6.1f} ms  pico en vuelo {pico:5}  CPU tirada {100 * cpu_tirada / max(cpu_total, 1e-9):5.1f} %")
   
   tasa = float(sys.argv[1]) if len(sys.argv) > 1 else 100
   print(f"{tasa:.0f} req/s contra un pod que termina ≈ 167 req/s (2 slots ÷ 12 ms de decode promedio)")
   for con_admision in (False, True):
       simular(tasa, con_admision)
   EOF
   python3.13 admision.py 100

   Qué deberías ver: Dos líneas con los mismos números: a tiempo 100.0 %, 503 0.0 %, p99 a tiempo 55.7 ms, pico en vuelo 9. Con ρ ≈ 0.6 (100 ÷ 167) las puertas no hacen nada: así debe verse el día a día.

2. Paso 2Copiar

   python3.13 admision.py 150

   Qué deberías ver: ρ ≈ 0.9. sin límite: todo a tiempo, pero el p99 sube a 111.6 ms y hay hasta 22 en vuelo. con admisión: 97.7 % a tiempo, 2.3 % de 503 y p99 de 73.8 ms, nunca más de 16. Cerca del límite, las puertas cambian un poco de rechazo por un p99 más bajo; los números finos (16, 50 ms) se ajustan con una prueba de carga (load test).

3. Paso 3Copiar

   python3.13 admision.py 250

   Qué deberías ver: 1.5 veces la capacidad. sin límite: a tiempo 2.8 %, tarde 97.2 %, pico en vuelo 1554 y CPU tirada 97.4 %: el pod trabaja al máximo para nadie y llega a guardar 1554 bodies a la vez: con fotos web de varios MB, eso termina en OOMKilled. con admisión: a tiempo 68.6 % (≈ 167 ÷ 250), 503 31.4 %, p99 de 90.2 ms, 16 en vuelo y nada de CPU tirada.

4. Paso 4Copiar

   cd ~ && rm -rf ~/lab-gateway

   Qué deberías ver: Sin salida.

Probado de verdad en un Mac. Ejecutado en macOS con Python 3.9.6, 3.13.16 y 3.14.3, y repetido copiando los pasos tal cual en zsh: salidas idénticas, con las cifras de arriba. Es una simulación de eventos, sin esperas reales: llegadas al azar (Poisson), decode de 4 ms más una exponencial de 8 ms (≈ 12 ms en promedio, ≈ 167 req/s con 2 slots), 8 ms de Triton y las puertas de main.py (16 en vuelo, 2 slots, 50 ms, deadline de 250 ms).

Qué vas a provocar: Ver, con un gateway mínimo y un upstream falso, qué código, qué headers y qué texto recibe el caller en cada falla, y qué deja ver debug=True.

Necesitas: Python 3.13 e internet para pip (fastapi, uvicorn y sus dependencias). Dos terminales. Ni Docker ni GPU.

1. Paso 1Copiar

   mkdir -p ~/lab-gateway && cd ~/lab-gateway
   [ -d .venv ] || python3.13 -m venv .venv
   . .venv/bin/activate
   pip install "fastapi==0.141.1" "uvicorn==0.54.0"
   cat > errores.py <<'EOF'
   # errores.py: un gateway mínimo con un upstream falso, para ver qué recibe el caller en cada falla
   import asyncio, os, uuid
   from fastapi import FastAPI, HTTPException, Request
   
   class Rechazo(Exception):  # hace el papel de InferenceServerException
       pass
   
   async def upstream(modo):  # un Triton falso
       if modo == "lento":
           await asyncio.sleep(1.0)
       if modo == "caido":
           raise ConnectionRefusedError("connect ECONNREFUSED 10.0.3.7:8000")
       if modo == "rechazo":
           raise Rechazo("Request timeout expired")
       if modo == "bug":
           return {}["logits"]  # un bug nuestro: KeyError
       return [0.83, 0.07, 0.04]
   
   app = FastAPI(debug=os.getenv("DEBUG") == "1")
   
   @app.post("/v1/classify")
   async def classify(request: Request, modo: str = "ok"):
       request_id = request.headers.get("x-request-id") or uuid.uuid4().hex
       try:
           try:
               scores = await asyncio.wait_for(upstream(modo), timeout=0.15)
           except TimeoutError:
               raise HTTPException(504, "timeout de inferencia") from None
           except ConnectionError:
               raise HTTPException(503, "upstream no disponible", headers={"Retry-After": "1"}) from None
           except Rechazo:
               raise HTTPException(503, "upstream rechazó la inferencia", headers={"Retry-After": "1"}) from None
           return {"scores": scores}
       except HTTPException as exc:
           headers = {**(exc.headers or {}), "X-Request-ID": request_id}
           raise HTTPException(exc.status_code, exc.detail, headers=headers) from None
   EOF

   Qué deberías ver: Successfully installed, con fastapi-0.141.1 y uvicorn-0.54.0 en la lista (Requirement already satisfied si ya los tenías). Son las versiones de C01; FastAPI saca versiones casi a diario (0.160.0 el 22 de octubre de 2026) y lo que se ve aquí no cambia.

2. Paso 2Copiar

   # en una segunda terminal
   cd ~/lab-gateway && . .venv/bin/activate
   uvicorn errores:app --port 8080

   Qué deberías ver: Uvicorn running on http://127.0.0.1:8080 (Press CTRL+C to quit). Déjala abierta: ahí salen los logs del servidor.

3. Paso 3Copiar

   for m in ok lento caido rechazo bug; do
     curl -s -o /dev/null -w "$m %{http_code} %{time_total}s\n" -X POST "localhost:8080/v1/classify?modo=$m"
   done

   Qué deberías ver: ok 200; lento 504 en ≈ 0.15 s, no en 1 s; caido 503; rechazo 503; bug 500. En la segunda terminal, cada uno deja su línea de access log (el bug, con 500 Internal Server Error) y el bug, además, ERROR: Exception in ASGI application y un traceback que termina en KeyError: 'logits'.

4. Paso 4Copiar

   curl -i -X POST "localhost:8080/v1/classify?modo=caido" -H "X-Request-ID: abc123"

   Qué deberías ver: HTTP/1.1 503 Service Unavailable, con retry-after: 1, x-request-id: abc123 y content-type: application/json; el body es {"detail":"upstream no disponible"}. El texto original (connect ECONNREFUSED 10.0.3.7:8000, con una IP interna) no llegó al caller.

5. Paso 5Copiar

   curl -s -X POST "localhost:8080/v1/classify?modo=bug"; echo

   Qué deberías ver: Internal Server Error, en texto plano: ni el KeyError, ni el archivo, ni la línea.

6. Paso 6Copiar

   # segunda terminal: Ctrl+C y vuelve a arrancar con debug
   DEBUG=1 uvicorn errores:app --port 8080
   # primera terminal
   curl -s -X POST "localhost:8080/v1/classify?modo=bug" | tail -4

   Qué deberías ver: Ahora la respuesta es el traceback completo, en texto porque curl no pide HTML: rutas de tus archivos, la línea return {}["logits"] y KeyError: 'logits' al final. Eso le regala debug=True a cualquiera que llame.

7. Paso 7Copiar

   # segunda terminal: Ctrl+C
   deactivate
   cd ~ && rm -rf ~/lab-gateway

   Qué deberías ver: Sin salida.

Verificado contra la documentación; no ejecutado aquí. Sin instalar nada: errores.py se corrió en macOS con FastAPI 0.143.0, Starlette 1.7.0 y uvicorn 0.54.0 (Python 3.13.16), en otro puerto, y dio estas salidas: lento respondió en 0.154 s; con DEBUG=1, tail -4 termina en KeyError: 'logits', y con Accept: text/html el traceback sale en HTML. Para la versión del C01 se verificó en el código: el http_exception_handler de FastAPI responde {"detail": …} con los headers del HTTPException; Starlette pasa los nombres de header a minúsculas y su ServerErrorMiddleware responde Internal Server Error en texto plano con debug=False, o el traceback con debug=True, y relanza la excepción para que uvicorn la loguee como Exception in ASGI application.

Qué vas a provocar: Ver con uvicorn de verdad que el lifespan corre antes de abrir el puerto, que liveness y readiness responden cosas distintas, y cómo drena (o corta) un request en curso al recibir SIGTERM.

Necesitas: Python 3.13 e internet para pip (fastapi y uvicorn). Dos terminales con zsh; curl y pkill vienen con macOS. Ni Docker ni GPU.

1. Paso 1Copiar

   mkdir -p ~/lab-gateway && cd ~/lab-gateway
   [ -d .venv ] || python3.13 -m venv .venv
   . .venv/bin/activate
   pip install "fastapi==0.141.1" "uvicorn==0.54.0"
   cat > salud.py <<'EOF'
   # salud.py: lifespan con warmup, liveness, readiness y un request largo para ver el apagado
   import asyncio, os
   from contextlib import asynccontextmanager
   from fastapi import FastAPI, HTTPException
   
   DEP = os.path.expanduser("~/lab-gateway/dependencia_lista")  # hace de «el modelo está listo en Triton»
   
   @asynccontextmanager
   async def lifespan(app: FastAPI):
       print("lifespan: creo los clientes y caliento", flush=True)
       await asyncio.sleep(float(os.getenv("WARMUP_S", "10")))  # carga y warmup simulados
       print("lifespan: listo", flush=True)
       yield
       print("lifespan: cierro el cliente del upstream", flush=True)
   
   app = FastAPI(lifespan=lifespan)
   
   @app.get("/healthz")
   async def healthz():
       return {"status": "ok"}  # liveness: sin dependencias
   
   @app.get("/readyz")
   async def readyz():
       if not os.path.exists(DEP):  # readiness: depende de la dependencia
           raise HTTPException(503, "la dependencia no está lista")
       return {"status": "ready"}
   
   @app.get("/lento")
   async def lento(s: float = 2.0):
       await asyncio.sleep(s)  # un request que tarda s segundos
       return {"durmio_s": s}
   EOF

   Qué deberías ver: Successfully installed, con fastapi-0.141.1 y uvicorn-0.54.0 en la lista (o Requirement already satisfied si ya estaban).

2. Paso 2Copiar

   # segunda terminal: el servidor
   cd ~/lab-gateway && . .venv/bin/activate
   uvicorn salud:app --port 8080 --timeout-graceful-shutdown 5
   # primera terminal, enseguida
   for i in $(seq 1 14); do curl -s -o /dev/null -w "%{http_code} " localhost:8080/healthz; sleep 1; done; echo

   Qué deberías ver: Primero varios 000 (connection refused: todavía nadie escucha en el puerto) y después 200. En la segunda terminal: Waiting for application startup., lifespan: creo los clientes y caliento, 10 s después lifespan: listo, Application startup complete. y recién entonces Uvicorn running on http://127.0.0.1:8080. Es lo que ve una probe durante el arranque.

3. Paso 3Copiar

   curl -s -w " %{http_code}\n" localhost:8080/readyz
   curl -s -w " %{http_code}\n" localhost:8080/healthz
   touch ~/lab-gateway/dependencia_lista
   curl -s -w " %{http_code}\n" localhost:8080/readyz

   Qué deberías ver: {"detail":"la dependencia no está lista"} 503, después {"status":"ok"} 200 y, con el archivo creado, {"status":"ready"} 200. Sin la dependencia, el proceso sigue vivo pero no recibiría tráfico: es lo que pasa con todos los gateways de C01 cuando Triton cae.

4. Paso 4Copiar

   curl -s -w " %{http_code}\n" "localhost:8080/lento?s=2" & sleep 0.5; pkill -TERM -f "uvicorn salud:app"; wait
   # en la segunda terminal, cuando uvicorn termine
   echo $?

   Qué deberías ver: ≈ 2 s después llega {"durmio_s":2.0} 200: el request en curso terminó aunque el SIGTERM llegó en medio. En la segunda terminal: Shutting down, Waiting for connections to close. (CTRL+C to force quit), la línea del GET /lento?s=2 con 200 OK, Waiting for application shutdown., lifespan: cierro el cliente del upstream, Application shutdown complete. y Finished server process [PID]. Luego zsh: terminated y echo $? da 143: uvicorn relanza el SIGTERM al terminar. Como PID 1 de un contenedor, ese relanzamiento se ignora y sale con 0 (hoja de Docker).

5. Paso 5Copiar

   # segunda terminal: sin warmup y con solo 3 s de drenado
   WARMUP_S=0 uvicorn salud:app --port 8080 --timeout-graceful-shutdown 3
   # primera terminal
   curl -s -w " %{http_code}\n" "localhost:8080/lento?s=10" & sleep 0.5; pkill -TERM -f "uvicorn salud:app"; wait

   Qué deberías ver: ≈ 3 s después del SIGTERM llega Internal Server Error 500. En la segunda terminal: ERROR: Cancel 1 running task(s), timeout graceful shutdown exceeded y un ERROR: Exception in ASGI application con un CancelledError, y la línea del GET /lento?s=10 con 500: uvicorn cancela el request, le manda un 500 y sigue con el apagado. En Kubernetes, lo que siga vivo al vencer la gracia muere con SIGKILL, sin siquiera ese 500.

6. Paso 6Copiar

   deactivate
   cd ~ && rm -rf ~/lab-gateway

   Qué deberías ver: Sin salida.

Verificado contra la documentación; no ejecutado aquí. Sin instalar nada: salud.py se corrió en macOS con FastAPI 0.143.0, Starlette 1.7.0 y uvicorn 0.54.0 (Python 3.13.16), en otro puerto y con dependencia_lista en otra carpeta, y dio estas salidas: 11 veces 000 y después 200; 503, 200 y 200; {"durmio_s":2.0} 200 con el SIGTERM en medio, los mensajes en el orden de arriba y salida 143; con --timeout-graceful-shutdown 3, Internal Server Error 500 ≈ 3.2 s después del SIGTERM. En el código de uvicorn 0.54.0: el lifespan corre antes de abrir el socket; una tarea cancelada antes de empezar su respuesta recibe un 500 (send_500_response); y al salir relanza la señal capturada con signal.raise_signal.