Día 5 -- Circuit breaker y documentación

1from datetime import datetime, timedelta
2from collections import defaultdict
3
4class GPSCircuitBreaker:
5    """
6    Circuit breaker para detectar y aislar GPSs defectuosos.
7    
8    Estados por driver:
9    - CLOSED: funcionamiento normal, eventos pasan
10    - OPEN: driver aislado, eventos descartados
11    - HALF_OPEN: período de prueba, si la tasa baja vuelve a CLOSED
12    
13    Umbrales:
14    - max_events_per_window: máximo eventos permitidos en la ventana
15    - window_seconds: tamaño de la ventana de conteo
16    - cooldown_seconds: tiempo que un driver permanece en OPEN
17    """
18    
19    def __init__(
20        self,
21        max_events_per_window: int = 10,  # 10 ev en 30s = ~0.33/s (normal es 0.2/s)
22        window_seconds: int = 30,
23        cooldown_seconds: int = 300,  # 5 minutos de cuarentena
24    ):
25        self.max_events = max_events_per_window
26        self.window_seconds = window_seconds
27        self.cooldown_seconds = cooldown_seconds
28        
29        # Estado por driver
30        self.event_counts: dict[str, list[datetime]] = defaultdict(list)
31        self.tripped_drivers: dict[str, datetime] = {}  # driver -> trip time
32        self.stats = {"passed": 0, "blocked": 0, "trips": 0}
33    
34    def should_allow(self, driver_id: str, event_time: datetime) -> bool:
35        """
36        Decide si un evento debe pasar o ser bloqueado.
37        Retorna True si el evento es aceptado, False si bloqueado.
38        """
39        # Si está en cuarentena, verificar si el cooldown expiro
40        if driver_id in self.tripped_drivers:
41            trip_time = self.tripped_drivers[driver_id]
42            if (event_time - trip_time).total_seconds() < self.cooldown_seconds:
43                self.stats["blocked"] += 1
44                return False  # Sigue en cuarentena
45            else:
46                # Cooldown expirado -> half-open -> reset
47                del self.tripped_drivers[driver_id]
48                self.event_counts[driver_id] = []
49        
50        # Limpiar eventos fuera de la ventana
51        window_start = event_time - timedelta(seconds=self.window_seconds)
52        self.event_counts[driver_id] = [
53            t for t in self.event_counts[driver_id] if t >= window_start
54        ]
55        
56        # Anadir evento actual
57        self.event_counts[driver_id].append(event_time)
58        
59        # Verificar si excede el umbral
60        if len(self.event_counts[driver_id]) > self.max_events:
61            # TRIP! Aislar este driver
62            self.tripped_drivers[driver_id] = event_time
63            self.stats["trips"] += 1
64            self.stats["blocked"] += 1
65            return False
66        
67        self.stats["passed"] += 1
68        return True
69    
70    def get_tripped_drivers(self) -> list[str]:
71        """Retorna lista de drivers actualmente en cuarentena."""
72        return list(self.tripped_drivers.keys())
73    
74    def get_stats(self) -> dict:
75        """Estadisticas del circuit breaker."""
76        return {
77            **self.stats,
78            "active_trips": len(self.tripped_drivers),
79            "block_rate_pct": (
80                self.stats["blocked"] / 
81                (self.stats["passed"] + self.stats["blocked"]) * 100
82                if (self.stats["passed"] + self.stats["blocked"]) > 0 else 0
83            ),
84        }

Circuit breaker: detecta GPSs con tasa anomala y los aisla automaticamente

1# Integrar circuit breaker en el pipeline de Spark
2
3# El CB se inicializa una vez en el driver
4circuit_breaker = GPSCircuitBreaker(
5    max_events_per_window=10,
6    window_seconds=30,
7    cooldown_seconds=300
8)
9
10def process_with_circuit_breaker(batch_df, batch_id):
11    """
12    foreachBatch que aplica el circuit breaker antes de procesar.
13    """
14    if batch_df.isEmpty():
15        return
16    
17    rows = batch_df.collect()
18    
19    # Filtrar con circuit breaker
20    allowed_rows = []
21    for row in rows:
22        event_time = row['event_time']
23        if circuit_breaker.should_allow(row['driver_id'], event_time):
24            allowed_rows.append(row)
25    
26    # Loggear si hay drivers aislados
27    tripped = circuit_breaker.get_tripped_drivers()
28    if tripped:
29        stats = circuit_breaker.get_stats()
30        print(f"[Batch {batch_id}] Circuit breaker: "
31              f"{len(tripped)} drivers aislados, "
32              f"{stats['block_rate_pct']:.1f}% eventos bloqueados")
33        # Alertar a Carlos para que investigue los dispositivos
34        # alert_slack(f"GPS defectuosos detectados: {tripped}")
35    
36    # Procesar solo eventos permitidos
37    if allowed_rows:
38        allowed_df = spark.createDataFrame(allowed_rows, batch_df.schema)
39        # ... calcular ETA, escribir a DynamoDB ...
40
41# Uso en el pipeline
42query = clean_gps_stream.writeStream \
43    .foreachBatch(process_with_circuit_breaker) \
44    .trigger(processingTime="30 seconds") \
45    .start()

Integración del circuit breaker en el foreachBatch de Spark

1# ADR-007: Pipeline de ETAs en tiempo real
2# Status: Accepted
3# Date: 2024-11-22
4# Author: tú
5
6## Contexto
7FlashRoute necesita ETAs actualizados cada 30 segundos para 1000+
8repartidores activos. Ana y su equipo de coordinadores toman
9decisiones de reasignacion basadas en estos datos.
10
11## Decision
12Implementar pipeline con Spark Structured Streaming:
13- Fuente: Kafka topic driver.location
14- Calculo: Formula Haversine + factor urbano 1.4x
15- Sink: DynamoDB (tabla delivery_etas, TTL 2h)
16- Trigger: cada 30 segundos (processingTime)
17- Proteccion: watermark 30s + dedup + circuit breaker
18
19## Alternativas consideradas
20
21## Consecuencias
22- Positivas: latencia <30s, escalable (probado 3x en Black Friday)
23- Negativas: complejidad operativa, requiere EMR cluster activo 24/7
24- Riesgos: single point of failure (mitigado con checkpoints)
25
26## Metricas de éxito
27- Latencia dashboard: < 30 segundos (p99)
28- Precision ETA: error < 5 minutos para distancias < 5km
29- Disponibilidad: 99.9% uptime (max 8 min downtime/semana)

ADR-007: documenta la arquitectura del pipeline de ETAs