Pruebas de Carga y Ajustes Finales

Concepto clave

Las pruebas de carga son evaluaciones que simulan múltiples usuarios concurrentes accediendo a tu API para medir su rendimiento bajo estrés. Imagina un restaurante que normalmente atiende 20 mesas, pero de repente recibe una reserva para 200 personas: las pruebas de carga te dicen si tu cocina (servidor) y meseros (endpoints) pueden manejar el pico sin colapsar.

En APIs de ML, esto es crítico porque los modelos suelen ser computacionalmente costosos. Un endpoint de predicción que responde en 100ms con un usuario puede volverse inservible con 100 usuarios simultáneos si no está optimizado. Los ajustes finales son modificaciones basadas en métricas como latencia, throughput y uso de recursos para garantizar que la API cumpla con los SLAs (Acuerdos de Nivel de Servicio) en producción.

Cómo funciona en la práctica

Primero, defines escenarios realistas: ¿cuántas peticiones por segundo esperas? ¿Cuál es el peor caso? Luego, usas herramientas como Locust o k6 para simular esos usuarios. Configuras un script que haga peticiones POST a tu endpoint de predicción con datos de prueba variados.

Ejemplo paso a paso:

1. Preparas un entorno de staging idéntico a producción.
2. Ejecutas una prueba de carga gradual: 10, 50, 100 usuarios en 5 minutos.
3. Mides métricas clave: tiempo de respuesta promedio, percentil 95, errores por segundo.
4. Identificas cuellos de botella: ¿es el modelo, la CPU, la memoria?
5. Aplicas ajustes: caching, batching, optimización de código.
6. Repites hasta cumplir los objetivos.

Código en acción

Script de prueba con Locust (antes del ajuste):

from locust import HttpUser, task, between
import json

class MLAPIUser(HttpUser):
    wait_time = between(1, 3)
    
    @task
    def predict(self):
        payload = {"features": [5.1, 3.5, 1.4, 0.2]}
        headers = {"Content-Type": "application/json"}
        self.client.post("/predict", json=payload, headers=headers)

Endpoint FastAPI inicial (lento):

from fastapi import FastAPI
import pickle
import numpy as np

app = FastAPI()
model = pickle.load(open("model.pkl", "rb"))

@app.post("/predict")
async def predict(features: list):
    # Carga el modelo en cada petición (ineficiente)
    prediction = model.predict([features])
    return {"prediction": prediction.tolist()}

Endpoint optimizado (después del ajuste):

from fastapi import FastAPI
from functools import lru_cache
import pickle
import numpy as np

app = FastAPI()

@lru_cache(maxsize=1)
def load_model():
    return pickle.load(open("model.pkl", "rb"))

@app.post("/predict")
async def predict(features: list):
    model = load_model()  # Modelo en caché
    # Batch processing para múltiples peticiones
    if isinstance(features[0], list):
        predictions = model.predict(features)
    else:
        predictions = model.predict([features])
    return {"prediction": predictions.tolist()}

Errores comunes

• Probar en local en vez de staging: Los recursos de tu laptop no reflejan producción. Usa un entorno con specs similares.
• Ignorar el percentil 95: El tiempo promedio puede engañar. Si el P95 es alto, algunos usuarios tendrán mala experiencia.
• No monitorear recursos del servidor: Una prueba puede pasar pero consumir 100% de CPU, lo que colapsará en horas.
• Usar datos de prueba irreales: Si pruebas con features simplificadas, no verás el costo real del preprocesamiento.
• Olvidar los timeouts: Configura timeouts en el cliente y servidor para evitar peticiones zombis.

Checklist de dominio

1. ¿Definiste objetivos de rendimiento (ej: <200ms de latencia con 100 RPS)?
2. ¿Ejecutaste pruebas con al menos 3 niveles de carga (baja, media, pico)?
3. ¿Implementaste caching para modelos o datos estáticos?
4. ¿Configuraste logging de métricas de rendimiento (ej: con Prometheus)?
5. ¿Validaste que el uso de memoria sea estable durante la prueba?
6. ¿Documentaste los límites conocidos de tu API?
7. ¿Probaste el recovery después de un pico de carga?

Optimiza un endpoint de ML bajo carga

Tu API tiene un endpoint /predict que usa un modelo de scikit-learn. En pruebas, con 50 usuarios concurrentes, el tiempo de respuesta sube a 2 segundos (objetivo: <500ms). Sigue estos pasos:

1. Clona el repositorio base: git clone https://github.com/ejemplo/ml-load-test.git
2. Ejecuta la prueba de carga inicial con locust -f test_load.py --host http://localhost:8000 y anota el P95.
3. Analiza el código en app.py: identifica 2 ineficiencias (ej: carga repetida del modelo).
4. Implementa optimizaciones: caching con @lru_cache, reducir serialización, o agregar batch processing.
5. Vuelve a ejecutar la prueba y verifica que el P95 baje a menos de 500ms.
6. Agrega métricas de monitorización usando prometheus-fastapi-instrumentator.

Entrega un reporte con: métricas antes/después, código modificado, y explicación de cada cambio.

• Revisa cómo se carga el modelo en cada petición. ¿Puede cachearse?
• El endpoint recibe un solo ejemplo. ¿Podría manejar batches para mejorar throughput?
• Usa herramientas como `py-spy` para profiling si no encuentras el cuello de botella.

Evalua tu comprension

Completa el quiz interactivo de arriba para ganar XP.