7. Drill-down y Correlación con Traces

Descripción de la cápsula

Tu dashboard muestra un spike de latencia a las 14:30. El p95 pasó de 1.2s a 4.8s. Lo ves claramente en el time series panel. Pero ahora necesitas responder: ¿por qué? ¿Qué requests específicos causaron el spike? ¿Fue un modelo? ¿Un endpoint? ¿Un tipo de prompt?

Sin correlación con traces, el dashboard es diagnóstico incompleto. Te dice "algo está mal" pero no "esto es lo que está mal". Necesitas el puente entre la vista agregada (métricas en Grafana) y la vista individual (traces en Jaeger). Ese puente se llama drill-down y la herramienta técnica que lo habilita se llama exemplars.

Un exemplar es un trace ID adjunto a un metric data point. Cuando Prometheus almacena un histograma de latencia, el exemplar dice "este data point de 4.8s corresponde al trace abc123". Grafana muestra los exemplars como puntos clicables en el time series. Haces click en uno y te lleva al trace en Jaeger donde puedes ver exactamente qué pasó en ese request.

Esta cápsula te enseña a configurar exemplars, vincular Grafana con Jaeger, y construir investigation paths — secuencias lógicas de drill-down que van de la métrica al diagnóstico.

Exemplars: El Puente entre Métricas y Traces

Qué es un exemplar

En Prometheus, un exemplar es metadata adicional asociada a un metric sample. Contiene:

• Un label traceID con el trace ID del request que generó ese data point
• Opcionalmente otros labels como spanID

# Ejemplo de un metric con exemplar en formato Prometheus
ai_request_duration_seconds_bucket{le="2.0",model="gpt-4o-mini"} 142
  # {traceID="4bf92f3577b34da6a3ce929d0e0e4736"} 1.847 1709900000.000

Esa línea dice: "este bucket de ≤2.0s tiene 142 requests, y uno de ellos tiene el trace ID 4bf92f... con un valor de 1.847s".

Configurar Exemplars en Python

Para que tu app envíe exemplars, necesitas conectar OpenTelemetry con prometheus_client:

import time
import random
from prometheus_client import Histogram, Counter, generate_latest, CONTENT_TYPE_LATEST
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from fastapi import FastAPI
from starlette.responses import Response

resource = Resource.create({
    "service.name": "ai-app-with-exemplars",
    "service.version": "1.0.0",
})

provider = TracerProvider(resource=resource)
exporter = OTLPSpanExporter(endpoint="http://localhost:4317", insecure=True)
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("ai-app", "1.0.0")

AI_REQUEST_DURATION = Histogram(
    "ai_request_duration_seconds",
    "Request duration with exemplars",
    ["model", "endpoint"],
    buckets=[0.1, 0.25, 0.5, 0.75, 1.0, 1.5, 2.0, 3.0, 5.0, 7.5, 10.0],
)

AI_REQUESTS = Counter(
    "ai_requests_total",
    "Total requests",
    ["model", "endpoint", "status"],
)

app = FastAPI()


def observe_with_exemplar(histogram, labels: dict, value: float):
    """Registra un valor en el histogram con el trace ID actual como exemplar."""
    current_span = trace.get_current_span()
    span_context = current_span.get_span_context()

    if span_context.is_valid:
        trace_id = format(span_context.trace_id, "032x")
        span_id = format(span_context.span_id, "016x")
        exemplar = {"traceID": trace_id, "spanID": span_id}
        histogram.labels(**labels).observe(value, exemplar)
    else:
        histogram.labels(**labels).observe(value)


@app.post("/api/ask")
async def ask():
    with tracer.start_as_current_span("api.ask") as span:
        model = random.choice(["gpt-4o-mini", "gpt-4o"])
        endpoint = "/api/ask"

        latency = random.gauss(1.0 if model == "gpt-4o-mini" else 2.0, 0.3)
        latency = max(0.1, latency)
        time.sleep(0.01)

        span.set_attribute("gen_ai.request.model", model)
        span.set_attribute("duration_ms", round(latency * 1000, 2))

        observe_with_exemplar(
            AI_REQUEST_DURATION,
            {"model": model, "endpoint": endpoint},
            latency,
        )

        AI_REQUESTS.labels(model=model, endpoint=endpoint, status="success").inc()

        return {"status": "ok", "model": model, "latency_ms": round(latency * 1000)}


@app.get("/metrics")
async def metrics():
    return Response(
        content=generate_latest(),
        media_type=CONTENT_TYPE_LATEST,
    )

La función observe_with_exemplar es la clave: obtiene el trace ID del span actual y lo adjunta como exemplar al histogram observation. Prometheus almacena este exemplar, y Grafana lo muestra como un punto clicable.

Configurar Grafana para Exemplars

Prerequisitos

1. Prometheus corriendo con --enable-feature=exemplar-storage (ya configurado en el docker-compose de la cápsula 03)
2. Jaeger configurado como data source en Grafana
3. Prometheus data source configurado con exemplar trace ID destination

Data Source: Prometheus con Exemplars

Si usaste el provisioning de la cápsula 03, ya está configurado. Si no, en Grafana UI:

1. Configuration → Data Sources → Prometheus
2. En la sección "Exemplars", habilita:
   • Internal link: enabled
   • Data source: Jaeger
   • Label name: traceID

# grafana/provisioning/datasources/datasources.yml
apiVersion: 1

datasources:
  - name: Prometheus
    type: prometheus
    access: proxy
    url: http://prometheus:9090
    isDefault: true
    jsonData:
      httpMethod: POST
      exemplarTraceIdDestinations:
        - name: traceID
          datasourceUid: jaeger
          urlDisplayLabel: "View in Jaeger"

  - name: Jaeger
    type: jaeger
    access: proxy
    url: http://jaeger:16686
    uid: jaeger

Verificar que Exemplars Funcionan

import httpx


def verify_exemplars():
    print("=" * 55)
    print("EXEMPLAR VERIFICATION")
    print("=" * 55)

    print("\n1. Generating test traffic...")
    for i in range(20):
        try:
            httpx.post("http://localhost:8000/api/ask", timeout=10.0)
        except Exception:
            pass
    print(f"   Sent 20 requests")

    print("\n2. Checking Prometheus for exemplars...")
    try:
        r = httpx.get(
            "http://localhost:9090/api/v1/query_exemplars",
            params={
                "query": "ai_request_duration_seconds_bucket",
                "start": "2024-01-01T00:00:00Z",
                "end": "2030-12-31T23:59:59Z",
            },
            timeout=10.0,
        )
        data = r.json()

        if data.get("status") == "success" and data.get("data"):
            exemplars = data["data"]
            total = sum(len(e.get("exemplars", [])) for e in exemplars)
            print(f"   Found {total} exemplars across {len(exemplars)} series")

            if exemplars:
                first = exemplars[0].get("exemplars", [{}])[0]
                labels = first.get("labels", {})
                trace_id = labels.get("traceID", "N/A")
                print(f"   Sample traceID: {trace_id}")
                print(f"   → Open in Jaeger: http://localhost:16686/trace/{trace_id}")
        else:
            print("   No exemplars found. Make sure exemplar-storage is enabled.")
    except Exception as e:
        print(f"   Error: {e}")

    print("\n3. Checking Jaeger for traces...")
    try:
        r = httpx.get(
            "http://localhost:16686/api/services",
            timeout=5.0,
        )
        services = r.json().get("data", [])
        if "ai-app-with-exemplars" in services:
            print(f"   Service found in Jaeger: ai-app-with-exemplars")
        else:
            print(f"   Available services: {services}")
    except Exception as e:
        print(f"   Error: {e}")

    print("\n" + "=" * 55)
    print("HOW TO TEST IN GRAFANA:")
    print("=" * 55)
    print("1. Open a latency time series panel")
    print("2. Enable 'Exemplars' toggle in the query options")
    print("3. You should see diamond-shaped points on the chart")
    print("4. Click one → it opens the trace in Jaeger")
    print("=" * 55)


verify_exemplars()

Investigation Paths: De Métrica a Diagnóstico

Un investigation path es una secuencia de pasos que sigues para ir de una anomalía en el dashboard al diagnóstico. No es ad-hoc — es un proceso repetible.

Path 1: Latency Spike Investigation

TRIGGER: P95 latency spike en el dashboard

PASO 1: ¿Cuándo empezó?
  → Dashboard time series: identificar el momento exacto
  → Narrow el rango temporal a ±15 minutos del spike

PASO 2: ¿Qué modelo/endpoint?
  → Panel "Latency by Model": ¿solo un modelo subió?
  → Panel "Latency by Endpoint": ¿solo un endpoint?
  → Esto reduce el espacio de búsqueda

PASO 3: ¿Request rate cambió?
  → Panel "Requests per Minute": ¿más tráfico?
  → Si sí: el spike es por carga
  → Si no: el spike es por algo diferente

PASO 4: Click en exemplar
  → En el time series de latencia, click en un punto del spike
  → Abre el trace en Jaeger
  → Ve qué paso del pipeline tardó más

PASO 5: Diagnóstico
  → El trace muestra: "vector_search tardó 8s"
  → O: "gen_ai.chat tardó 6s, gen_ai.request.model=gpt-4o"
  → O: "prompt.estimated_tokens=3200, se estaba enviando mucho contexto"

Path 2: Cost Spike Investigation

TRIGGER: Cost per hour superó el budget line

PASO 1: ¿Cuánto y desde cuándo?
  → Panel "Cost per Hour": identificar la magnitud y duración
  → ¿Es un spike puntual o un incremento sostenido?

PASO 2: ¿Qué modelo?
  → Panel "Cost by Model": ¿todo el incremento es de gpt-4o?
  → Si sí: alguien cambió un endpoint a gpt-4o

PASO 3: ¿Token usage cambió?
  → Panel "Token Usage": ¿prompt tokens subieron?
  → Si sí: se está enviando más contexto
  → Si no: más requests, no más tokens per request

PASO 4: Click en exemplar del período caro
  → Abre un trace de un request costoso
  → Ve: prompt_tokens=2800, más del doble de lo normal
  → Ve: prompt.context_docs=8, normalmente son 3

PASO 5: Diagnóstico
  → El RAG retrieval cambió de top_k=3 a top_k=8
  → Más documentos = más tokens = más costo

Path 3: Quality Degradation Investigation

TRIGGER: Quality score trend bajando

PASO 1: ¿Gradual o repentino?
  → Panel "Quality Score Trend": ¿bajó de golpe o lleva días bajando?
  → Gradual: drift en datos o modelo. Repentino: cambio de configuración.

PASO 2: ¿Qué tipo de errores funcionales?
  → Panel "Error Distribution": ¿hallucinations subieron?
  → Panel "Error Rate by Type": ¿cuándo empezaron a subir?

PASO 3: ¿Modelo específico?
  → Quality score by model: ¿solo gpt-4o-mini degradó?
  → Si sí: el proveedor puede haber actualizado el modelo

PASO 4: Muestrear traces del período degradado
  → Abrir traces con exemplars del rango temporal
  → Comparar respuestas de ahora vs respuestas de hace una semana
  → Verificar si el sistema prompt o el contexto RAG cambiaron

PASO 5: Diagnóstico
  → El proveedor actualizó gpt-4o-mini-2024-07-18 a una nueva versión
  → O: el system prompt fue editado y perdió una instrucción importante

Código: Investigation Path Runner

from datetime import datetime, timedelta


def build_investigation_context(
    anomaly_type: str,
    time_start: str,
    time_end: str,
    affected_metric: str,
) -> dict:
    promql_queries = {}

    promql_queries["primary_metric"] = {
        "latency_spike": f"histogram_quantile(0.95, sum(rate(ai_request_duration_seconds_bucket[5m])) by (le))",
        "cost_spike": "sum(rate(ai_cost_usd_total[1h])) * 3600",
        "quality_degradation": "avg(avg_over_time(ai_quality_score[1h]))",
        "error_spike": "sum(rate(ai_errors_total[5m])) / sum(rate(ai_requests_total[5m])) * 100",
    }.get(anomaly_type, "up")

    promql_queries["breakdown_by_model"] = {
        "latency_spike": "histogram_quantile(0.95, sum(rate(ai_request_duration_seconds_bucket[5m])) by (le, model))",
        "cost_spike": "sum(rate(ai_cost_usd_total[1h])) by (model) * 3600",
        "quality_degradation": "avg(avg_over_time(ai_quality_score[1h])) by (model)",
        "error_spike": "sum(rate(ai_errors_total[5m])) by (type) / sum(rate(ai_requests_total[5m])) * 100",
    }.get(anomaly_type, "up")

    promql_queries["request_rate"] = "sum(rate(ai_requests_total[5m])) * 60"

    promql_queries["exemplar_query"] = {
        "latency_spike": "ai_request_duration_seconds_bucket",
        "cost_spike": "ai_cost_usd_total",
        "quality_degradation": "ai_quality_score",
        "error_spike": "ai_errors_total",
    }.get(anomaly_type, "")

    investigation_steps = {
        "latency_spike": [
            "1. Identify spike timing in the latency time series",
            "2. Check latency breakdown by model — is one model responsible?",
            "3. Check request rate — did traffic increase?",
            "4. Click exemplar on the spike to open trace in Jaeger",
            "5. In Jaeger: identify which span is the bottleneck",
            "6. Check span attributes: tokens, model, context_docs",
        ],
        "cost_spike": [
            "1. Check cost per hour panel — spike vs sustained increase",
            "2. Check cost by model — which model is driving cost",
            "3. Check token usage — prompt or completion tokens increased?",
            "4. Check cost per request — did individual requests get more expensive?",
            "5. Click exemplar on expensive period to see trace",
            "6. In trace: check prompt_tokens, context_docs, model used",
        ],
        "quality_degradation": [
            "1. Check quality trend — gradual or sudden drop?",
            "2. Check error distribution — hallucination rate increased?",
            "3. Check quality by model — specific model degraded?",
            "4. Compare responses from before vs after degradation",
            "5. Check if system prompt or RAG config changed",
            "6. Check if model version changed (provider update)",
        ],
        "error_spike": [
            "1. Check error rate — technical or functional errors?",
            "2. Check error distribution pie chart — which type dominates?",
            "3. Check latency — if both up, likely infrastructure issue",
            "4. Check provider status page for outages",
            "5. Click exemplar on error period to see failing traces",
            "6. In trace: check error status and exception details",
        ],
    }

    return {
        "anomaly_type": anomaly_type,
        "time_range": {"start": time_start, "end": time_end},
        "queries": promql_queries,
        "steps": investigation_steps.get(anomaly_type, []),
        "jaeger_search_url": f"http://localhost:16686/search?service=ai-app&start={time_start}&end={time_end}",
        "grafana_explore_url": f"http://localhost:3000/explore?left=%5B%22now-6h%22,%22now%22%5D",
    }


investigation = build_investigation_context(
    anomaly_type="latency_spike",
    time_start="2026-03-08T14:00:00Z",
    time_end="2026-03-08T15:00:00Z",
    affected_metric="ai_request_duration_seconds",
)

print("INVESTIGATION CONTEXT")
print("=" * 55)
print(f"Type:       {investigation['anomaly_type']}")
print(f"Time range: {investigation['time_range']['start']} — {investigation['time_range']['end']}")
print(f"\nPromQL Queries:")
for name, query in investigation["queries"].items():
    print(f"  {name}: {query[:70]}...")
print(f"\nInvestigation Steps:")
for step in investigation["steps"]:
    print(f"  {step}")
print(f"\nJaeger: {investigation['jaeger_search_url']}")
print(f"Grafana: {investigation['grafana_explore_url']}")

Grafana → Jaeger: Panel con Exemplar Links

Para habilitar exemplars en un panel existente, agrega exemplar: true a la query:

import json

panel_with_exemplars = {
    "title": "E2E Latency with Exemplars",
    "type": "timeseries",
    "gridPos": {"h": 10, "w": 24, "x": 0, "y": 0},
    "fieldConfig": {
        "defaults": {
            "unit": "s",
            "custom": {
                "lineWidth": 2,
                "fillOpacity": 10,
                "showPoints": "never",
            },
        },
    },
    "targets": [
        {
            "refId": "A",
            "expr": "histogram_quantile(0.95, sum(rate(ai_request_duration_seconds_bucket[5m])) by (le))",
            "legendFormat": "p95",
            "exemplar": True,
        },
        {
            "refId": "B",
            "expr": "histogram_quantile(0.50, sum(rate(ai_request_duration_seconds_bucket[5m])) by (le))",
            "legendFormat": "p50",
            "exemplar": True,
        },
    ],
    "options": {
        "tooltip": {"mode": "multi"},
        "legend": {"displayMode": "table", "placement": "bottom", "calcs": ["mean", "max"]},
    },
}

print(json.dumps(panel_with_exemplars, indent=2))
print()
print("KEY: 'exemplar: true' in targets enables clickable trace links")
print("When you see diamond shapes on the chart, click them to open Jaeger")

En Grafana, los exemplars aparecen como diamantes pequeños en el time series. Cada diamante es un request específico. Haces click, se abre Jaeger con el trace completo.

Ejercicios

Ejercicio 1: Configurar Exemplars End-to-End (Fácil)

Escribe una app FastAPI mínima que: (1) crea spans con OpenTelemetry, (2) registra métricas en Prometheus con exemplars, (3) expone /metrics. Verifica que los exemplars aparecen en Prometheus.

from fastapi import FastAPI
from prometheus_client import Histogram

import time
import random
from fastapi import FastAPI
from prometheus_client import Histogram, generate_latest, CONTENT_TYPE_LATEST
from starlette.responses import Response
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
from opentelemetry.sdk.resources import Resource

resource = Resource.create({"service.name": "exemplar-test"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("exemplar-test")

DURATION = Histogram(
    "test_request_duration_seconds",
    "Test request duration",
    ["endpoint"],
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0],
)

app = FastAPI()


@app.get("/test")
async def test_endpoint():
    with tracer.start_as_current_span("test.request") as span:
        duration = random.gauss(1.0, 0.3)
        duration = max(0.05, duration)

        span.set_attribute("test.duration_ms", round(duration * 1000, 2))

        ctx = trace.get_current_span().get_span_context()
        trace_id = format(ctx.trace_id, "032x")
        span_id = format(ctx.span_id, "016x")

        DURATION.labels(endpoint="/test").observe(
            duration,
            {"traceID": trace_id, "spanID": span_id},
        )

        return {
            "duration_ms": round(duration * 1000),
            "trace_id": trace_id,
        }


@app.get("/metrics")
async def metrics():
    return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8001)

Ejercicio 2: Construir un Investigation Playbook (Medio)

Escribe una función que, dada una anomalía detectada, genere un playbook completo de investigación con: queries PromQL específicas, URLs directas a Grafana y Jaeger, y un checklist de pasos.

anomaly = {
    "type": "cost_spike",
    "detected_at": "2026-03-08T14:30:00Z",
    "severity": "high",
    "metric_value": 2.50,
    "normal_value": 0.40,
}

from datetime import datetime, timedelta


def generate_playbook(anomaly: dict) -> dict:
    detected = datetime.fromisoformat(anomaly["detected_at"].replace("Z", "+00:00"))
    window_start = (detected - timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ")
    window_end = (detected + timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ")
    ratio = anomaly["metric_value"] / anomaly["normal_value"]

    playbooks = {
        "cost_spike": {
            "title": "Cost Spike Investigation",
            "severity": anomaly["severity"],
            "summary": f"Cost spiked to ${anomaly['metric_value']}/hr "
                       f"({ratio:.1f}x normal ${anomaly['normal_value']}/hr)",
            "queries": {
                "cost_hourly": "sum(rate(ai_cost_usd_total[1h])) * 3600",
                "cost_by_model": "sum(rate(ai_cost_usd_total[1h])) by (model) * 3600",
                "tokens_by_model": "sum(rate(ai_tokens_total[5m])) by (model, type) * 60",
                "cost_per_request": "sum(increase(ai_cost_usd_total[1h])) / sum(increase(ai_requests_total[1h]))",
                "request_rate": "sum(rate(ai_requests_total[5m])) * 60",
            },
            "urls": {
                "grafana_cost_dashboard": "http://localhost:3000/d/cost-dashboard?from=now-3h&to=now",
                "jaeger_traces": f"http://localhost:16686/search?service=ai-app&start={window_start}&end={window_end}",
                "prometheus_explorer": "http://localhost:9090/graph?g0.expr=sum(rate(ai_cost_usd_total[1h]))*3600",
            },
            "checklist": [
                {"step": "Verify cost spike in Cost per Hour panel", "status": "pending"},
                {"step": "Check Cost by Model — which model drove the spike?", "status": "pending"},
                {"step": "Check Token Usage — did prompt or completion tokens increase?", "status": "pending"},
                {"step": "Check Cost per Request — individual requests got more expensive?", "status": "pending"},
                {"step": "Check Request Rate — more requests or same volume?", "status": "pending"},
                {"step": "Click exemplar in spike window to open trace", "status": "pending"},
                {"step": "In trace: check gen_ai.request.model, prompt_tokens", "status": "pending"},
                {"step": "Identify root cause and document", "status": "pending"},
            ],
            "common_causes": [
                "Model change: endpoint switched from gpt-4o-mini to gpt-4o",
                "Prompt expansion: more context docs included (RAG top_k increased)",
                "Traffic spike: sudden increase in request volume",
                "Retry storm: errors causing excessive retries",
                "Long prompts: user sending very long inputs",
            ],
        },
    }

    playbook = playbooks.get(anomaly["type"], {
        "title": "Unknown Anomaly Investigation",
        "summary": "No playbook defined for this anomaly type",
        "checklist": [{"step": "Manual investigation required", "status": "pending"}],
    })

    playbook["anomaly"] = anomaly
    playbook["time_window"] = {"start": window_start, "end": window_end}

    return playbook


anomaly = {
    "type": "cost_spike",
    "detected_at": "2026-03-08T14:30:00Z",
    "severity": "high",
    "metric_value": 2.50,
    "normal_value": 0.40,
}

playbook = generate_playbook(anomaly)

print(f"{'=' * 60}")
print(f"INVESTIGATION PLAYBOOK: {playbook['title']}")
print(f"{'=' * 60}")
print(f"Severity: {playbook['severity']}")
print(f"Summary:  {playbook['summary']}")
print(f"Window:   {playbook['time_window']['start']} — {playbook['time_window']['end']}")
print(f"\nQueries:")
for name, query in playbook.get("queries", {}).items():
    print(f"  {name}: {query[:65]}...")
print(f"\nURLs:")
for name, url in playbook.get("urls", {}).items():
    print(f"  {name}: {url}")
print(f"\nChecklist:")
for item in playbook["checklist"]:
    print(f"  [ ] {item['step']}")
print(f"\nCommon Causes:")
for cause in playbook.get("common_causes", []):
    print(f"  - {cause}")

Ejercicio 3: Simular un Incident y Trazar (Medio)

Escribe un script que simule un incident (latency spike durante 5 minutos), genere traces durante el incident, y después genere el investigation context con links a Jaeger.

import time
import httpx

import time
import httpx
from datetime import datetime


def simulate_incident(
    app_url: str = "http://localhost:8000",
    normal_requests: int = 50,
    incident_requests: int = 30,
    post_incident_requests: int = 50,
):
    print("=" * 55)
    print("INCIDENT SIMULATION")
    print("=" * 55)

    trace_ids = {"normal": [], "incident": [], "recovery": []}

    print("\n[Phase 1] Normal traffic...")
    for i in range(normal_requests):
        try:
            r = httpx.post(f"{app_url}/simulate?requests=5", timeout=30.0)
        except Exception:
            pass
    print(f"  Sent {normal_requests} batches of normal traffic")

    incident_start = datetime.now().isoformat()
    print(f"\n[Phase 2] INCIDENT — Starting at {incident_start}")
    print("  Simulating high-latency traffic...")

    for i in range(incident_requests):
        try:
            r = httpx.post(f"{app_url}/simulate?requests=20", timeout=30.0)
        except Exception:
            pass
        time.sleep(0.5)
    incident_end = datetime.now().isoformat()
    print(f"  Incident ended at {incident_end}")

    print(f"\n[Phase 3] Recovery traffic...")
    for i in range(post_incident_requests):
        try:
            r = httpx.post(f"{app_url}/simulate?requests=5", timeout=30.0)
        except Exception:
            pass
    print(f"  Sent {post_incident_requests} batches of recovery traffic")

    print(f"\n{'=' * 55}")
    print("INVESTIGATION LINKS")
    print(f"{'=' * 55}")
    print(f"\nIncident window: {incident_start} — {incident_end}")
    print(f"\nGrafana Dashboard:")
    print(f"  http://localhost:3000/d/ai-ops?from=now-30m&to=now")
    print(f"\nJaeger (all traces during incident):")
    print(f"  http://localhost:16686/search?service=ai-app&start={incident_start}&end={incident_end}")
    print(f"\nPrometheus (latency during incident):")
    print(f"  http://localhost:9090/graph?g0.expr=histogram_quantile(0.95,sum(rate(ai_request_duration_seconds_bucket[5m]))by(le))")
    print(f"\nInvestigation steps:")
    print(f"  1. Open Grafana → verify spike in latency panel")
    print(f"  2. Check which model/endpoint spiked")
    print(f"  3. Click exemplar in the spike → open trace in Jaeger")
    print(f"  4. In Jaeger: identify bottleneck span")
    print(f"  5. Compare with a trace from normal period")


simulate_incident()

Ejercicio 4: Comparar Traces de Normal vs Incident (Avanzado)

Escribe una función que, dados dos sets de trace IDs (normal period y incident period), compare los atributos de los spans y identifique las diferencias. ¿El modelo cambió? ¿Los tokens subieron? ¿Un componente tardó mucho más?

normal_traces = []
incident_traces = []

import random
import statistics


def generate_mock_traces(n: int, period: str) -> list[dict]:
    traces = []
    for i in range(n):
        if period == "normal":
            model = random.choices(["gpt-4o-mini", "gpt-4o"], weights=[0.8, 0.2])[0]
            prompt_tokens = random.randint(300, 600)
            completion_tokens = random.randint(100, 300)
            embed_ms = random.gauss(200, 50)
            search_ms = random.gauss(500, 100)
            llm_ms = random.gauss(800 if model == "gpt-4o-mini" else 1500, 200)
            context_docs = 3
        else:
            model = random.choices(["gpt-4o-mini", "gpt-4o"], weights=[0.5, 0.5])[0]
            prompt_tokens = random.randint(600, 1500)
            completion_tokens = random.randint(200, 500)
            embed_ms = random.gauss(300, 80)
            search_ms = random.gauss(1200, 300)
            llm_ms = random.gauss(1500 if model == "gpt-4o-mini" else 3000, 500)
            context_docs = random.randint(5, 8)

        total_ms = embed_ms + search_ms + llm_ms

        traces.append({
            "trace_id": f"trace_{period}_{i:04d}",
            "model": model,
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "embed_ms": max(50, round(embed_ms)),
            "search_ms": max(100, round(search_ms)),
            "llm_ms": max(200, round(llm_ms)),
            "total_ms": max(500, round(total_ms)),
            "context_docs": context_docs,
        })
    return traces


def compare_trace_sets(normal: list[dict], incident: list[dict]) -> dict:
    def aggregate(traces: list[dict]) -> dict:
        return {
            "count": len(traces),
            "model_distribution": {
                model: sum(1 for t in traces if t["model"] == model)
                for model in set(t["model"] for t in traces)
            },
            "avg_prompt_tokens": round(statistics.mean(t["prompt_tokens"] for t in traces)),
            "avg_completion_tokens": round(statistics.mean(t["completion_tokens"] for t in traces)),
            "avg_embed_ms": round(statistics.mean(t["embed_ms"] for t in traces)),
            "avg_search_ms": round(statistics.mean(t["search_ms"] for t in traces)),
            "avg_llm_ms": round(statistics.mean(t["llm_ms"] for t in traces)),
            "avg_total_ms": round(statistics.mean(t["total_ms"] for t in traces)),
            "p95_total_ms": round(sorted(t["total_ms"] for t in traces)[int(len(traces) * 0.95)]),
            "avg_context_docs": round(statistics.mean(t["context_docs"] for t in traces), 1),
        }

    normal_stats = aggregate(normal)
    incident_stats = aggregate(incident)

    diffs = {}
    for key in normal_stats:
        if key in ("count", "model_distribution"):
            continue
        n_val = normal_stats[key]
        i_val = incident_stats[key]
        if n_val > 0:
            change_pct = (i_val - n_val) / n_val * 100
        else:
            change_pct = 0
        diffs[key] = {
            "normal": n_val,
            "incident": i_val,
            "change_pct": round(change_pct, 1),
            "significant": abs(change_pct) > 30,
        }

    root_causes = []
    if diffs["avg_search_ms"]["change_pct"] > 50:
        root_causes.append(f"Vector search slowed by {diffs['avg_search_ms']['change_pct']:.0f}%")
    if diffs["avg_llm_ms"]["change_pct"] > 50:
        root_causes.append(f"LLM call slowed by {diffs['avg_llm_ms']['change_pct']:.0f}%")
    if diffs["avg_prompt_tokens"]["change_pct"] > 30:
        root_causes.append(f"Prompt tokens increased by {diffs['avg_prompt_tokens']['change_pct']:.0f}%")
    if diffs["avg_context_docs"]["change_pct"] > 30:
        root_causes.append(f"Context docs increased from {diffs['avg_context_docs']['normal']} to {diffs['avg_context_docs']['incident']}")

    n_dist = normal_stats["model_distribution"]
    i_dist = incident_stats["model_distribution"]
    gpt4o_normal_pct = n_dist.get("gpt-4o", 0) / normal_stats["count"] * 100
    gpt4o_incident_pct = i_dist.get("gpt-4o", 0) / incident_stats["count"] * 100
    if gpt4o_incident_pct > gpt4o_normal_pct + 15:
        root_causes.append(
            f"gpt-4o usage increased from {gpt4o_normal_pct:.0f}% to {gpt4o_incident_pct:.0f}%"
        )

    return {
        "normal_stats": normal_stats,
        "incident_stats": incident_stats,
        "diffs": diffs,
        "root_causes": root_causes,
    }


normal_traces = generate_mock_traces(50, "normal")
incident_traces = generate_mock_traces(50, "incident")

comparison = compare_trace_sets(normal_traces, incident_traces)

print("TRACE COMPARISON: Normal vs Incident")
print("=" * 65)
print(f"\n{'Metric':25s} {'Normal':>10s} {'Incident':>10s} {'Change':>10s} {'Flag':>6s}")
print("-" * 65)
for metric, data in comparison["diffs"].items():
    flag = " ***" if data["significant"] else ""
    print(
        f"{metric:25s} {data['normal']:>10} {data['incident']:>10} "
        f"{data['change_pct']:>+9.1f}%{flag}"
    )

print(f"\nProbable Root Causes:")
for cause in comparison["root_causes"]:
    print(f"  - {cause}")

Resumen

• Exemplars son el puente entre métricas (vista agregada) y traces (vista individual). Un trace ID adjunto a un metric data point te permite ir de "p95 está alto" a "este request específico tardó 4.8s y así se ve su trace".
• Configurar exemplars requiere tres cosas: Prometheus con --enable-feature=exemplar-storage, tu app enviando exemplars con trace IDs, y Grafana configurado para mostrar exemplars y linkar a Jaeger.
• Investigation paths son secuencias repetibles para ir de anomalía a diagnóstico: spike en dashboard → narrow time range → identify model/endpoint → click exemplar → analyze trace → root cause.
• Playbooks de investigación documentan estos paths para cada tipo de anomalía (latency spike, cost spike, quality degradation, error spike). En producción, se integran con incident management.
• La comparación normal vs incident revela las diferencias: ¿cambió el modelo? ¿Subieron los tokens? ¿Un componente se degradó? Las métricas con >30% cambio son los probable root causes.
• El drill-down cierra el loop de observabilidad: instrumentas → visualizas → investigas → diagnosticas. Sin drill-down, el loop se rompe en "visualizas" y nunca llegas a "diagnosticas".

Recursos Adicionales

1. Prometheus — Exemplars — Documentación de exemplars en Prometheus
2. Grafana — Exemplars — Cómo configurar y visualizar exemplars en Grafana
3. Grafana — Data Source Links — Vincular Prometheus con Jaeger via exemplars
4. OpenTelemetry — Exemplars — Spec de exemplars en OTel
5. Jaeger — Search API — API de búsqueda para automatizar investigación
6. Incident Response — Google SRE Book — Prácticas de respuesta a incidentes