4. Validación con Incidentes Simulados

Descripción

Esta es la cápsula más importante del módulo. Tienes el pipeline configurado y verificado link por link. Pero un pipeline que "funciona" en condiciones normales no garantiza que funciona cuando lo necesitas — durante un incidente. La única forma de saberlo es simular incidentes reales y verificar que el stack responde end-to-end: la anomalía se genera → el trace la captura → la métrica la refleja → el dashboard la muestra → la alerta dispara → el runbook es ejecutable con los datos disponibles.

Vas a simular 3 incidentes: un cost spike, un hallucination spike, y una latency degradation. Cada simulación inyecta el problema en tu app y verifica que cada capa del stack lo detecta. Si alguna capa falla — si la alerta no dispara, si el dashboard no muestra el spike, si el trace no tiene los atributos que el runbook necesita — lo descubres ahora, no a las 3am con usuarios afectados.

Conexión con la cápsula anterior: La cápsula 03 verificó que el pipeline funciona link por link en condiciones normales. Esta cápsula lo somete a estrés: ¿funciona cuando las métricas se salen de lo normal? ¿Las alertas detectan anomalías? ¿Los traces contienen lo que el debugging necesita?

Por Qué Simular Incidentes

El gap de confianza

LO QUE SABES HASTA AHORA                LO QUE NO SABES
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
✅ Los traces se exportan               ¿Los traces capturan anomalías?
✅ Las métricas se scrape               ¿Las métricas reflejan spikes?
✅ Los dashboards muestran datos         ¿Los dashboards muestran EL spike?
✅ Las alert rules existen               ¿Las alertas DISPARAN cuando deben?
✅ Los runbooks están escritos           ¿Los runbooks son ejecutables con datos reales?

Simular incidentes cierra ese gap. Después de esta cápsula, no "crees" que tu stack funciona — lo sabes.

Los 3 incidentes

Incidente                Qué simulas                    Qué validas
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
1. Cost Spike            Requests con tokens 10x         Alert de cost dispara
                         mayores de lo normal             Dashboard muestra spike
                                                         Trace identifica el endpoint

2. Hallucination Spike   Requests donde                  Alert de hallucination dispara
                         hallucination_score > 0.7        Dashboard muestra degradación
                                                         Trace tiene score para debugging

3. Latency Degradation   Requests con latencia            Alert de latency SLO dispara
                         5x mayor de lo normal            Dashboard muestra bottleneck
                                                         Trace identifica el span lento

El IncidentSimulator

import time
import random
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum

from opentelemetry import trace
from opentelemetry.trace import SpanKind


class IncidentType(Enum):
    COST_SPIKE = "cost_spike"
    HALLUCINATION_SPIKE = "hallucination_spike"
    LATENCY_DEGRADATION = "latency_degradation"


@dataclass
class IncidentConfig:
    incident_type: IncidentType
    duration_seconds: int
    intensity: float  # 1.0 = normal, 5.0 = 5x peor
    affected_endpoint: str = "/api/chat"
    affected_model: str = "gpt-4o"
    requests_per_second: float = 2.0


@dataclass
class IncidentResult:
    incident_type: IncidentType
    started_at: str
    ended_at: str
    total_requests: int
    trace_ids: list[str]
    anomalous_values: list[dict]
    detection_expected: dict


class IncidentSimulator:
    """Simula incidentes para validación end-to-end del observability stack."""

    def __init__(self, tracer: trace.Tracer):
        self.tracer = tracer
        self.results: list[IncidentResult] = []

    def _generate_normal_request(self) -> dict:
        return {
            "prompt_tokens": random.randint(100, 300),
            "completion_tokens": random.randint(50, 150),
            "latency_ms": random.uniform(200, 800),
            "hallucination_score": random.uniform(0.0, 0.15),
            "cost_usd": random.uniform(0.001, 0.005),
        }

    def _generate_cost_spike_request(self, intensity: float) -> dict:
        base = self._generate_normal_request()
        base["prompt_tokens"] = int(base["prompt_tokens"] * intensity)
        base["completion_tokens"] = int(base["completion_tokens"] * intensity)
        base["cost_usd"] = base["cost_usd"] * intensity * intensity
        return base

    def _generate_hallucination_request(self, intensity: float) -> dict:
        base = self._generate_normal_request()
        base["hallucination_score"] = min(0.5 + (intensity * 0.1), 0.95)
        return base

    def _generate_latency_request(self, intensity: float) -> dict:
        base = self._generate_normal_request()
        base["latency_ms"] = base["latency_ms"] * intensity
        return base

    def _instrument_request(self, data: dict, endpoint: str,
                            model: str, incident_tag: str) -> str:
        with self.tracer.start_as_current_span(
            "chat.request", kind=SpanKind.SERVER,
        ) as root_span:
            root_span.set_attribute("http.route", endpoint)
            root_span.set_attribute("incident.simulated", True)
            root_span.set_attribute("incident.type", incident_tag)

            time.sleep(data["latency_ms"] / 1000 * 0.01)

            with self.tracer.start_as_current_span("retrieval.query") as ret_span:
                ret_span.set_attribute("retrieval.num_results", random.randint(3, 8))
                ret_span.set_attribute("retrieval.relevance_score", random.uniform(0.4, 0.9))

            with self.tracer.start_as_current_span(
                "llm.completion", kind=SpanKind.CLIENT,
            ) as llm_span:
                llm_span.set_attribute("gen_ai.system", "openai")
                llm_span.set_attribute("gen_ai.request.model", model)
                llm_span.set_attribute("gen_ai.usage.prompt_tokens", data["prompt_tokens"])
                llm_span.set_attribute("gen_ai.usage.completion_tokens", data["completion_tokens"])
                llm_span.set_attribute("llm.cost_usd", data["cost_usd"])

            with self.tracer.start_as_current_span("quality.check") as q_span:
                q_span.set_attribute("llm.hallucination_score", data["hallucination_score"])
                q_span.set_attribute("llm.quality_score", 1.0 - data["hallucination_score"])

            return format(root_span.get_span_context().trace_id, "032x")

    def simulate_incident(self, config: IncidentConfig) -> IncidentResult:
        generators = {
            IncidentType.COST_SPIKE: self._generate_cost_spike_request,
            IncidentType.HALLUCINATION_SPIKE: self._generate_hallucination_request,
            IncidentType.LATENCY_DEGRADATION: self._generate_latency_request,
        }

        generator = generators[config.incident_type]
        start_time = datetime.now(timezone.utc).isoformat()
        trace_ids = []
        anomalous_values = []

        print(f"\n{'='*60}")
        print(f"SIMULATING: {config.incident_type.value}")
        print(f"  Duration: {config.duration_seconds}s | Intensity: {config.intensity}x")
        print(f"  Endpoint: {config.affected_endpoint} | RPS: {config.requests_per_second}")
        print(f"{'='*60}")

        end_time_target = time.time() + config.duration_seconds
        request_count = 0

        while time.time() < end_time_target:
            data = generator(config.intensity)
            trace_id = self._instrument_request(
                data, config.affected_endpoint,
                config.affected_model, config.incident_type.value,
            )
            trace_ids.append(trace_id)
            anomalous_values.append(data)
            request_count += 1

            if request_count % 5 == 0:
                print(f"  Sent {request_count} requests...")

            time.sleep(1.0 / config.requests_per_second)

        end_time = datetime.now(timezone.utc).isoformat()
        detection = self._expected_detections(config)

        result = IncidentResult(
            incident_type=config.incident_type,
            started_at=start_time, ended_at=end_time,
            total_requests=request_count, trace_ids=trace_ids,
            anomalous_values=anomalous_values, detection_expected=detection,
        )
        self.results.append(result)

        print(f"\n✅ Complete: {request_count} requests, {len(trace_ids)} traces")
        for k, v in detection.items():
            print(f"  Expected: {k} → {v}")

        return result

    def _expected_detections(self, config: IncidentConfig) -> dict:
        base = {"traces_in_jaeger": True, "dashboard_visible": True}

        if config.incident_type == IncidentType.COST_SPIKE:
            base["alert"] = "cost_spike should fire"
            base["metric"] = "cost_usd_total rate > 3x baseline"
            base["dashboard_panel"] = "Cost Tracking → USD per minute"
            base["runbook"] = "cost_spike.md → filter by high token count"

        elif config.incident_type == IncidentType.HALLUCINATION_SPIKE:
            base["alert"] = "hallucination_rate should fire"
            base["metric"] = "hallucination_score avg > 0.08"
            base["dashboard_panel"] = "AI Monitoring → Hallucination Rate"
            base["runbook"] = "hallucination_spike.md → filter by score > 0.7"

        elif config.incident_type == IncidentType.LATENCY_DEGRADATION:
            base["alert"] = "latency_slo should fire"
            base["metric"] = "request_duration_seconds p95 > SLO"
            base["dashboard_panel"] = "Latency Analysis → P95 over time"
            base["runbook"] = "latency_degradation.md → identify slowest span"

        return base

    def simulate_all_incidents(self) -> list[IncidentResult]:
        configs = [
            IncidentConfig(IncidentType.COST_SPIKE, duration_seconds=30,
                           intensity=8.0, requests_per_second=2.0),
            IncidentConfig(IncidentType.HALLUCINATION_SPIKE, duration_seconds=30,
                           intensity=5.0, requests_per_second=2.0),
            IncidentConfig(IncidentType.LATENCY_DEGRADATION, duration_seconds=30,
                           intensity=6.0, requests_per_second=1.0),
        ]

        print("\n" + "=" * 60)
        print("FULL INCIDENT SIMULATION SUITE")
        print("=" * 60)

        results = []
        for i, config in enumerate(configs, 1):
            print(f"\n--- Incident {i}/3 ---")
            results.append(self.simulate_incident(config))
            print("\nWaiting 10s between incidents...")
            time.sleep(10)

        self._print_summary(results)
        return results

    def _print_summary(self, results: list[IncidentResult]) -> None:
        print("\n" + "=" * 60)
        print("SIMULATION SUMMARY")
        print("=" * 60)
        for r in results:
            print(f"\n  {r.incident_type.value}: {r.total_requests} requests, "
                  f"{len(r.trace_ids)} traces")

Verificación de Detección

El DetectionVerifier

Después de simular, verifica que el stack detectó cada incidente:

import json
import urllib.request
import urllib.parse
from dataclasses import dataclass


@dataclass
class DetectionCheck:
    incident_type: str
    check_name: str
    passed: bool
    details: str


class DetectionVerifier:
    """Verifica que el stack detectó los incidentes simulados."""

    def __init__(self, jaeger_url: str = "http://localhost:16686",
                 prometheus_url: str = "http://localhost:9090"):
        self.jaeger_url = jaeger_url
        self.prometheus_url = prometheus_url
        self.checks: list[DetectionCheck] = []

    def verify_traces_captured(self, result: IncidentResult,
                               service: str = "ai-chat-service") -> DetectionCheck:
        try:
            tag = f"incident.type%3D{result.incident_type.value}"
            url = f"{self.jaeger_url}/api/traces?service={service}&tags={tag}&limit=100"
            req = urllib.request.Request(url)
            with urllib.request.urlopen(req, timeout=10) as resp:
                data = json.loads(resp.read())
            traces = data.get("data", [])
            check = DetectionCheck(
                result.incident_type.value, "traces_captured",
                len(traces) > 0,
                f"Found {len(traces)} traces (expected ~{result.total_requests})",
            )
        except Exception as e:
            check = DetectionCheck(result.incident_type.value, "traces_captured",
                                   False, f"Error: {e}")
        self.checks.append(check)
        return check

    def verify_metric_spike(self, result: IncidentResult) -> DetectionCheck:
        queries = {
            IncidentType.COST_SPIKE.value: "rate(cost_usd_total[5m])",
            IncidentType.HALLUCINATION_SPIKE.value: "avg(hallucination_score)",
            IncidentType.LATENCY_DEGRADATION.value:
                "histogram_quantile(0.95, rate(request_duration_seconds_bucket[5m]))",
        }
        query = queries.get(result.incident_type.value)
        try:
            url = f"{self.prometheus_url}/api/v1/query?query={urllib.parse.quote(query)}"
            req = urllib.request.Request(url)
            with urllib.request.urlopen(req, timeout=10) as resp:
                data = json.loads(resp.read())
            metric_results = data.get("data", {}).get("result", [])
            if metric_results:
                value = float(metric_results[0]["value"][1])
                check = DetectionCheck(result.incident_type.value, "metric_spike",
                                       True, f"Value: {value:.4f}")
            else:
                check = DetectionCheck(result.incident_type.value, "metric_spike",
                                       False, "No metric data")
        except Exception as e:
            check = DetectionCheck(result.incident_type.value, "metric_spike",
                                   False, f"Error: {e}")
        self.checks.append(check)
        return check

    def verify_alert_fired(self, result: IncidentResult) -> DetectionCheck:
        alert_names = {
            IncidentType.COST_SPIKE.value: "CostSpike",
            IncidentType.HALLUCINATION_SPIKE.value: "HallucinationRate",
            IncidentType.LATENCY_DEGRADATION.value: "LatencySLOBreach",
        }
        expected = alert_names.get(result.incident_type.value, "Unknown")
        try:
            url = f"{self.prometheus_url}/api/v1/alerts"
            req = urllib.request.Request(url)
            with urllib.request.urlopen(req, timeout=10) as resp:
                data = json.loads(resp.read())
            alerts = data.get("data", {}).get("alerts", [])
            firing = [a for a in alerts if a.get("state") == "firing"
                      and expected.lower() in a.get("labels", {}).get("alertname", "").lower()]
            check = DetectionCheck(
                result.incident_type.value, "alert_fired",
                len(firing) > 0,
                f"'{expected}': {'FIRING ✅' if firing else 'NOT FIRING — check threshold'}",
            )
        except Exception as e:
            check = DetectionCheck(result.incident_type.value, "alert_fired",
                                   False, f"Error: {e}")
        self.checks.append(check)
        return check

    def verify_trace_attributes(self, result: IncidentResult) -> DetectionCheck:
        required_by_type = {
            IncidentType.COST_SPIKE.value: [
                "gen_ai.usage.prompt_tokens", "gen_ai.usage.completion_tokens", "llm.cost_usd"],
            IncidentType.HALLUCINATION_SPIKE.value: [
                "llm.hallucination_score", "retrieval.relevance_score"],
            IncidentType.LATENCY_DEGRADATION.value: ["gen_ai.request.model"],
        }
        required = required_by_type.get(result.incident_type.value, [])
        if not required or not result.trace_ids:
            check = DetectionCheck(result.incident_type.value, "trace_attributes",
                                   False, "No traces or requirements")
            self.checks.append(check)
            return check
        try:
            url = f"{self.jaeger_url}/api/traces/{result.trace_ids[0]}"
            req = urllib.request.Request(url)
            with urllib.request.urlopen(req, timeout=10) as resp:
                data = json.loads(resp.read())
            all_tags = set()
            for span in data.get("data", [{}])[0].get("spans", []):
                for tag in span.get("tags", []):
                    all_tags.add(tag["key"])
            missing = [a for a in required if a not in all_tags]
            check = DetectionCheck(
                result.incident_type.value, "trace_attributes",
                len(missing) == 0,
                f"All present" if not missing else f"Missing: {missing}",
            )
        except Exception as e:
            check = DetectionCheck(result.incident_type.value, "trace_attributes",
                                   False, f"Error: {e}")
        self.checks.append(check)
        return check

    def verify_all(self, results: list[IncidentResult]) -> dict:
        print("\n" + "=" * 60)
        print("INCIDENT DETECTION VERIFICATION")
        print("=" * 60)

        all_results = {}
        for result in results:
            print(f"\n  {result.incident_type.value}:")
            checks = {
                "traces": self.verify_traces_captured(result),
                "metrics": self.verify_metric_spike(result),
                "alerts": self.verify_alert_fired(result),
                "attributes": self.verify_trace_attributes(result),
            }
            for name, check in checks.items():
                icon = "✅" if check.passed else "❌"
                print(f"    {icon} {check.check_name}: {check.details}")
            all_results[result.incident_type.value] = checks

        passed = sum(1 for c in self.checks if c.passed)
        total = len(self.checks)
        print(f"\n{'='*60}")
        print(f"TOTAL: {passed}/{total} checks passed")
        print("✅ STACK VALIDATED" if passed == total else f"⚠️  {total-passed} failures")
        return all_results


# --- Ejecutar simulación completa ---
# tracer = setup_tracing()
# simulator = IncidentSimulator(tracer)
# results = simulator.simulate_all_incidents()
# time.sleep(20)
# verifier = DetectionVerifier()
# verification = verifier.verify_all(results)

Interpretación de Resultados

Qué hacer cuando un check falla

Check fallido                    Causa probable                    Acción
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
traces_captured: FAIL            OTLP exporter no configurado      Verificar endpoint/puerto
                                 Jaeger no running                 docker-compose up jaeger

metric_spike: FAIL               Prometheus no scrapeando app      Verificar targets
                                 Métrica no existe                 Verificar instrumentación M2

alert_fired: FAIL                Threshold demasiado alto          Bajar threshold o subir intensity
                                 Alert rule no configurada         Verificar prometheus rules
                                 Evaluación > duración simulación  Esperar más entre sim y check

trace_attributes: FAIL           Span no registra el atributo      Agregar set_attribute en M3/M6
                                 Nombre diferente al esperado      Verificar naming conventions

Ejercicios

Ejercicio 1: Simular incidente de error rate (Fácil)

Agrega un cuarto tipo de incidente al IncidentSimulator: ERROR_SPIKE. Debe generar requests que producen errores HTTP 500. Configúralo para generar un 30% de error rate durante 30 segundos.

def generate_error_request(self, intensity: float) -> dict:
    base = self._generate_normal_request()
    error_probability = min(intensity * 0.06, 0.5)
    is_error = random.random() < error_probability
    if is_error:
        base["status_code"] = 500
        base["error_type"] = random.choice(["timeout", "rate_limit", "model_error"])
        base["completion_tokens"] = 0
    else:
        base["status_code"] = 200
        base["error_type"] = None
    return base


def instrument_error_request(self, data: dict, endpoint: str, model: str) -> str:
    with self.tracer.start_as_current_span(
        "chat.request", kind=SpanKind.SERVER,
    ) as root_span:
        root_span.set_attribute("http.route", endpoint)
        root_span.set_attribute("http.status_code", data["status_code"])
        root_span.set_attribute("incident.simulated", True)
        root_span.set_attribute("incident.type", "error_spike")

        if data["status_code"] == 500:
            root_span.set_status(trace.StatusCode.ERROR, data["error_type"])
            root_span.set_attribute("error.type", data["error_type"])

        REQUEST_TOTAL.labels(endpoint=endpoint, status=str(data["status_code"])).inc()
        return format(root_span.get_span_context().trace_id, "032x")

Ejercicio 2: Validar que el runbook es ejecutable (Medio)

Escribe una función que dado un IncidentResult, verifique paso a paso que los datos que el runbook necesita están disponibles: la query de Prometheus retorna datos, los atributos del trace existen.

@dataclass
class RunbookStep:
    step_number: int
    description: str
    check_type: str  # "query" o "trace_attribute"
    check_value: str
    passed: bool = False
    result: str = ""


def validate_runbook(result: IncidentResult, prometheus_url: str,
                     jaeger_url: str) -> list[RunbookStep]:
    runbook_steps = {
        IncidentType.COST_SPIKE.value: [
            RunbookStep(1, "Verificar cost rate", "query", "rate(cost_usd_total[5m])"),
            RunbookStep(2, "Verificar cost_usd en trace", "trace_attribute", "llm.cost_usd"),
            RunbookStep(3, "Verificar tokens en trace", "trace_attribute",
                        "gen_ai.usage.prompt_tokens"),
        ],
        IncidentType.HALLUCINATION_SPIKE.value: [
            RunbookStep(1, "Verificar hallucination rate", "query",
                        "avg(hallucination_score)"),
            RunbookStep(2, "Verificar score en trace", "trace_attribute",
                        "llm.hallucination_score"),
            RunbookStep(3, "Verificar relevance en trace", "trace_attribute",
                        "retrieval.relevance_score"),
        ],
        IncidentType.LATENCY_DEGRADATION.value: [
            RunbookStep(1, "Verificar latency p95", "query",
                        "histogram_quantile(0.95, rate(request_duration_seconds_bucket[5m]))"),
            RunbookStep(2, "Verificar modelo en trace", "trace_attribute",
                        "gen_ai.request.model"),
        ],
    }

    steps = runbook_steps.get(result.incident_type.value, [])
    print(f"\nRUNBOOK VALIDATION: {result.incident_type.value}")
    print("=" * 50)

    for step in steps:
        if step.check_type == "query":
            try:
                url = f"{prometheus_url}/api/v1/query?query={urllib.parse.quote(step.check_value)}"
                req = urllib.request.Request(url)
                with urllib.request.urlopen(req, timeout=10) as resp:
                    data = json.loads(resp.read())
                step.passed = len(data.get("data", {}).get("result", [])) > 0
                step.result = "Data available" if step.passed else "No data"
            except Exception as e:
                step.result = str(e)
        elif step.check_type == "trace_attribute" and result.trace_ids:
            try:
                url = f"{jaeger_url}/api/traces/{result.trace_ids[0]}"
                req = urllib.request.Request(url)
                with urllib.request.urlopen(req, timeout=10) as resp:
                    data = json.loads(resp.read())
                tags = set()
                for span in data.get("data", [{}])[0].get("spans", []):
                    for tag in span.get("tags", []):
                        tags.add(tag["key"])
                step.passed = step.check_value in tags
                step.result = "Found" if step.passed else "MISSING"
            except Exception as e:
                step.result = str(e)

        icon = "✅" if step.passed else "❌"
        print(f"  {icon} Step {step.step_number}: {step.description} → {step.result}")

    all_ok = all(s.passed for s in steps)
    print(f"\n{'✅ Runbook executable' if all_ok else '❌ Runbook has gaps'}")
    return steps

Ejercicio 3: Generar reporte de simulación en markdown (Medio)

Escribe una función que genere un reporte markdown con: resumen ejecutivo, detalle por incidente, y resultados de verificación.

def generate_simulation_report(results: list[IncidentResult],
                               verification: dict) -> str:
    lines = [
        "# Incident Simulation Report",
        f"\n**Generated:** {datetime.now(timezone.utc).isoformat()}",
        "",
        "## Executive Summary",
        f"- **Incidents simulated:** {len(results)}",
        f"- **Total requests:** {sum(r.total_requests for r in results)}",
        f"- **Total traces:** {sum(len(r.trace_ids) for r in results)}",
        "",
    ]

    for r in results:
        lines.extend([
            f"## {r.incident_type.value.replace('_', ' ').title()}",
            f"- **Requests:** {r.total_requests}",
            f"- **Traces:** {len(r.trace_ids)}",
            f"- **Period:** {r.started_at} → {r.ended_at}",
            "",
            "### Expected Detections",
        ])
        for k, v in r.detection_expected.items():
            lines.append(f"- **{k}:** {v}")

        if r.incident_type.value in verification:
            lines.append("\n### Verification")
            for name, check in verification[r.incident_type.value].items():
                icon = "✅" if check.passed else "❌"
                lines.append(f"- {icon} **{check.check_name}:** {check.details}")
        lines.append("")

    report = "\n".join(lines)
    print(report)
    return report

Ejercicio 4: Simulación con baseline comparison (Difícil)

Extiende el IncidentSimulator para que antes de simular, genere 20 requests normales como baseline. Después compara métricas normales vs incidente y calcula el factor de desviación.

@dataclass
class BaselineComparison:
    metric: str
    baseline_avg: float
    incident_avg: float
    deviation_factor: float
    anomalous: bool


def simulate_with_baseline(simulator: IncidentSimulator,
                           config: IncidentConfig,
                           baseline_requests: int = 20) -> tuple:
    print("\n--- Phase 1: Generating baseline ---")
    baseline_data = []
    for _ in range(baseline_requests):
        data = simulator._generate_normal_request()
        baseline_data.append(data)
        simulator._instrument_request(data, config.affected_endpoint,
                                       config.affected_model, "baseline")
        time.sleep(0.2)
    print(f"  Baseline: {baseline_requests} requests")

    print("\n--- Phase 2: Simulating incident ---")
    result = simulator.simulate_incident(config)

    comparisons = []
    for metric in ["prompt_tokens", "completion_tokens", "cost_usd",
                   "hallucination_score", "latency_ms"]:
        b_vals = [d[metric] for d in baseline_data if metric in d]
        i_vals = [d[metric] for d in result.anomalous_values if metric in d]
        if b_vals and i_vals:
            b_avg = sum(b_vals) / len(b_vals)
            i_avg = sum(i_vals) / len(i_vals)
            dev = i_avg / b_avg if b_avg > 0 else float("inf")
            comparisons.append(BaselineComparison(
                metric, b_avg, i_avg, dev, dev > 2.0 or dev < 0.5))

    print("\n--- Baseline Comparison ---")
    for c in comparisons:
        icon = "🔴" if c.anomalous else "🟢"
        print(f"  {icon} {c.metric}: baseline={c.baseline_avg:.4f}, "
              f"incident={c.incident_avg:.4f}, deviation={c.deviation_factor:.1f}x")
    return result, comparisons

Conexión con el Proyecto

La validación con incidentes simulados es el clímax del módulo:

1. Ejecutar los 3 incidentes simulados contra tu stack real
2. Verificar detección end-to-end con el DetectionVerifier
3. Validar que los runbooks son ejecutables con datos de las simulaciones
4. Documentar los resultados con el reporte de simulación

Lo que llevas al proyecto

VALIDACIÓN CON INCIDENTES — Checklist para el proyecto
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
☐ Cost spike simulado y detectado (alerta + dashboard + trace)
☐ Hallucination spike simulado y detectado
☐ Latency degradation simulada y detectada
☐ Traces tienen atributos que los runbooks necesitan
☐ Runbooks son ejecutables con datos reales
☐ Reporte de simulación generado y documentado
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

Resumen

• Simular incidentes es la validación definitiva del stack. Un pipeline que "funciona" en condiciones normales puede fallar cuando lo necesitas.
• Los 3 incidentes clave — cost spike, hallucination spike, latency degradation — cada uno prueba diferentes partes del stack.
• El IncidentSimulator inyecta anomalías controladas y registra trace IDs para verificación posterior.
• El DetectionVerifier confirma que cada capa detectó la anomalía: traces capturados, métricas reflejadas, alertas disparadas, atributos presentes.
• Validar los runbooks con datos reales es tan importante como verificar las alertas. Un runbook que pide datos que no existen es ficción operativa.
• La comparación con baseline es cómo funcionan los sistemas reales — detección de anomalías relativas, no thresholds absolutos.

Recursos Adicionales

1. Google SRE Book — Testing Reliability — Principios de testing de sistemas
2. Chaos Engineering — Principles — Framework para validación bajo estrés
3. Gremlin — Chaos Engineering Guide — Herramientas de chaos engineering
4. Prometheus — Unit Testing Rules — Testing de alerting rules
5. OpenTelemetry — Testing — Testing de instrumentación