Patrón Bulkhead: Aísla Recursos y Limita el Radio de Impacto

Resumen

El patrón bulkhead aísla recursos para que un fallo en una parte del sistema no se propague a las demás. Como los compartimentos estancos de un barco: si uno se inunda, el barco sigue a flote. En software, esto significa asignar thread pools, pools de conexión o límites de memoria separados a distintos servicios u operaciones. Si el servicio A agota su thread pool, el servicio B sigue teniendo hilos disponibles. Sin bulkheads, un único servicio lento aguas abajo puede consumir todos los hilos de un pool compartido y dejar sin recursos al resto de operaciones.

Cuando el pool de pagos se satura, solo se rechazan las llamadas de pago: inventario y notificaciones siguen funcionando porque su capacidad nunca fue compartida.

Cuándo Usarlo

• Dos o más servicios aguas abajo compartiendo un pool de recursos común (hilos, conexiones)
• Evitar que un servicio lento deje sin recursos al resto de operaciones
• Microservicios donde distintos endpoints tienen requisitos de SLA diferentes
• Aislamiento de conexiones a base de datos entre rutas críticas y no críticas
• Limitar el consumo de recursos por tenant o por servicio

Cuándo NO Usarlo

• Aplicaciones de un solo servicio sin dependencias aguas abajo
• Cuando el costo de mantener pools separados supera el beneficio del aislamiento
• Sistemas de bajo tráfico donde la contención es improbable
• Cuando el servicio aguas abajo es extremadamente fiable y rápido

Solución

Bulkhead con thread pool (Python)

# bulkhead/thread_pool.py — Aislamiento de thread pools por servicio
import concurrent.futures
from collections import defaultdict
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class BulkheadManager:
    """Administra thread pools separados para distintos servicios.
    Cada servicio tiene su propio pool: un servicio no puede agotar
    los hilos de otro."""

    def __init__(self):
        self._pools = {}
        self._configs = {}
        self._active = defaultdict(int)

    def configure(self, service_name, max_workers, queue_capacity=100):
        """Configura un bulkhead para un servicio específico."""
        self._pools[service_name] = concurrent.futures.ThreadPoolExecutor(
            max_workers=max_workers,
            thread_name_prefix=f"bulkhead-{service_name}"
        )
        self._configs[service_name] = {
            "max_workers": max_workers,
            "queue_capacity": queue_capacity
        }
        logger.info(f"Bulkhead configurado para {service_name}: "
                    f"max_workers={max_workers}, queue={queue_capacity}")

    def submit(self, service_name, fn, *args, **kwargs):
        """Envía una tarea al bulkhead de un servicio.
        Lanza BulkheadFullError si el pool y la cola están agotados."""
        if service_name not in self._pools:
            raise ValueError(f"No hay bulkhead configurado para {service_name}")

        config = self._configs[service_name]
        if self._active[service_name] >= config["max_workers"] + config["queue_capacity"]:
            raise BulkheadFullError(
                f"Bulkhead '{service_name}' lleno: "
                f"{self._active[service_name]} tareas activas"
            )

        self._active[service_name] += 1

        def wrapped_fn(*a, **kw):
            try:
                return fn(*a, **kw)
            finally:
                self._active[service_name] -= 1

        return self._pools[service_name].submit(wrapped_fn, *args, **kwargs)

    def get_stats(self):
        return {
            service: {
                "active": self._active[service],
                "max_workers": self._configs[service]["max_workers"],
                "queue_capacity": self._configs[service]["queue_capacity"],
                "utilization": self._active[service] / self._configs[service]["max_workers"]
            }
            for service in self._pools
        }

    def shutdown(self):
        for pool in self._pools.values():
            pool.shutdown(wait=True)


class BulkheadFullError(Exception):
    """Se lanza cuando el thread pool y la cola de un bulkhead están llenos."""
    pass


# Uso
bulkhead = BulkheadManager()
bulkhead.configure("payment", max_workers=10, queue_capacity=20)
bulkhead.configure("inventory", max_workers=5, queue_capacity=10)
bulkhead.configure("notifications", max_workers=3, queue_capacity=5)

def call_payment_api(amount):
    time.sleep(0.5)
    return {"status": "charged", "amount": amount}

# Payment tiene su propio pool: problemas en inventory no lo afectan
future = bulkhead.submit("payment", call_payment_api, 99.99)
result = future.result(timeout=5)

Bulkhead con semáforo (Python)

# bulkhead/semaphore.py — Límite de concurrencia con semáforos
import threading
import time
from contextlib import contextmanager
from functools import wraps

class SemaphoreBulkhead:
    """Limita las llamadas concurrentes usando un semáforo.
    Más liviano que un thread pool: funciona con cualquier modelo de ejecución."""

    def __init__(self, max_concurrent, timeout=30):
        self._semaphore = threading.Semaphore(max_concurrent)
        self._max = max_concurrent
        self._timeout = timeout
        self._active = 0
        self._lock = threading.Lock()

    @contextmanager
    def acquire(self):
        """Context manager que adquiere un permiso."""
        acquired = self._semaphore.acquire(timeout=self._timeout)
        if not acquired:
            # from bulkhead.thread_pool import BulkheadFullError
            raise BulkheadFullError(
                f"No se pudo adquirir el permiso en {self._timeout}s"
            )
        with self._lock:
            self._active += 1
        try:
            yield
        finally:
            with self._lock:
                self._active -= 1
            self._semaphore.release()

    def __call__(self, fn):
        """Decorador que envuelve una función con protección bulkhead."""
        @wraps(fn)
        def wrapper(*args, **kwargs):
            with self.acquire():
                return fn(*args, **kwargs)
        return wrapper

    @property
    def active(self):
        return self._active

    @property
    def available(self):
        return self._max - self._active


# Uso: proteger el acceso a la base de datos
db_bulkhead = SemaphoreBulkhead(max_concurrent=20)

@db_bulkhead
def query_database(sql):
    time.sleep(0.1)
    return {"rows": 42}

# Uso: proteger llamadas a APIs externas
api_bulkhead = SemaphoreBulkhead(max_concurrent=5, timeout=10)

def call_external_api(endpoint):
    with api_bulkhead.acquire():
        time.sleep(0.5)
        return {"data": "response"}

Bulkhead en Java con Spring

// BulkheadConfig.java — Bulkhead en Java usando resilience4j
import io.github.resilience4j.bulkhead.Bulkhead;
import io.github.resilience4j.bulkhead.BulkheadConfig;
import io.github.resilience4j.bulkhead.ThreadPoolBulkhead;
import io.github.resilience4j.bulkhead.ThreadPoolBulkheadConfig;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;

public class ServiceBulkheads {

    // Bulkhead basado en semáforo para llamadas síncronas
    private final Bulkhead paymentBulkhead;
    private final Bulkhead inventoryBulkhead;

    // Bulkhead con thread pool para llamadas asíncronas
    private final ThreadPoolBulkhead asyncPaymentBulkhead;

    public ServiceBulkheads() {
        // Payment: máx. 10 concurrentes, 30s de espera
        BulkheadConfig paymentConfig = BulkheadConfig.custom()
            .maxConcurrentCalls(10)
            .maxWaitDuration(Duration.ofSeconds(30))
            .build();
        this.paymentBulkhead = Bulkhead.of("payment", paymentConfig);

        // Inventory: máx. 5 concurrentes, 10s de espera
        BulkheadConfig inventoryConfig = BulkheadConfig.custom()
            .maxConcurrentCalls(5)
            .maxWaitDuration(Duration.ofSeconds(10))
            .build();
        this.inventoryBulkhead = Bulkhead.of("inventory", inventoryConfig);

        // Payment asíncrono: thread pool separado
        ThreadPoolBulkheadConfig asyncConfig = ThreadPoolBulkheadConfig.custom()
            .maxThreadPoolSize(10)
            .coreThreadPoolSize(5)
            .queueCapacity(20)
            .build();
        this.asyncPaymentBulkhead = ThreadPoolBulkhead.of("asyncPayment", asyncConfig);
    }

    public String callPayment(String orderId) {
        return Bulkhead.decorateSupplier(paymentBulkhead, () -> {
            // Llamada a la API de pagos
            return paymentClient.charge(orderId);
        }).get();
    }

    public String callInventory(String productId) {
        return Bulkhead.decorateSupplier(inventoryBulkhead, () -> {
            // Llamada a la API de inventario
            return inventoryClient.checkStock(productId);
        }).get();
    }

    public CompletableFuture<String> callPaymentAsync(String orderId) {
        return asyncPaymentBulkhead.executeSupplier(() ->
            paymentClient.charge(orderId)
        );
    }
}

Aislamiento de pools de conexión (Python)

# bulkhead/connection_pools.py — Pools de conexión separados por servicio
import psycopg2
from psycopg2 import pool
from contextlib import contextmanager

class IsolatedConnectionPools:
    """Mantiene pools de conexión a base de datos separados por servicio.
    Los servicios críticos no pueden quedarse sin conexiones por culpa
    de los no críticos."""

    def __init__(self):
        self._pools = {}

    def configure(self, service_name, db_url, min_conn, max_conn):
        self._pools[service_name] = pool.ThreadedConnectionPool(
            minconn=min_conn,
            maxconn=max_conn,
            dsn=db_url
        )

    @contextmanager
    def get_connection(self, service_name):
        if service_name not in self._pools:
            raise ValueError(f"No hay pool para {service_name}")

        conn = self._pools[service_name].getconn()
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            self._pools[service_name].putconn(conn)

    def close_all(self):
        for p in self._pools.values():
            p.closeall()


# Uso: servicios críticos y no críticos tienen pools separados
pools = IsolatedConnectionPools()

# Payment: crítico, recibe más conexiones
pools.configure("payment", "postgresql://db:5432/shop", min_conn=5, max_conn=20)

# Analytics: no crítico, menos conexiones
pools.configure("analytics", "postgresql://db:5432/shop", min_conn=2, max_conn=5)

def process_payment(order_id):
    with pools.get_connection("payment") as conn:
        cur = conn.cursor()
        cur.execute("UPDATE orders SET status='paid' WHERE id=%s", (order_id,))
        return cur.rowcount

def log_analytics(event):
    with pools.get_connection("analytics") as conn:
        cur = conn.cursor()
        cur.execute("INSERT INTO analytics_events (event) VALUES (%s)", (event,))

Cuotas de recursos en Kubernetes como bulkheads

# k8s-bulkhead.yaml — Límites de recursos como bulkheads en Kubernetes
apiVersion: v1
kind: ResourceQuota
metadata:
  name: critical-services-quota
  namespace: production
spec:
  hard:
    requests.cpu: "4"
    requests.memory: 8Gi
    limits.cpu: "8"
    limits.memory: 16Gi
    pods: "20"
---
apiVersion: v1
kind: ResourceQuota
metadata:
  name: batch-jobs-quota
  namespace: batch
spec:
  hard:
    requests.cpu: "2"
    requests.memory: 4Gi
    limits.cpu: "4"
    limits.memory: 8Gi
    pods: "10"
---
# Deployment con límites de recursos: este pod no puede consumir más que su límite
apiVersion: apps/v1
kind: Deployment
metadata:
  name: api-server
  namespace: production
spec:
  replicas: 3
  selector:
    matchLabels:
      app: api-server
  template:
    metadata:
      labels:
        app: api-server
    spec:
      containers:
        - name: api
          image: shop/api-server:latest
          resources:
            requests:
              cpu: "500m"
              memory: "512Mi"
            limits:
              cpu: "1000m"
              memory: "1Gi"

Cómo Funciona

Un bulkhead pone un techo duro a cuánto de un recurso compartido puede consumir una carga de trabajo. Eliges el recurso (hilos, conexiones, llamadas concurrentes, memoria), lo particionas por servicio o tenant, y haces que cada partición rechace el trabajo que excede en lugar de pedir prestado a sus vecinas. Tres propiedades lo hacen funcionar:

1. Capacidad acotada — todo bulkhead tiene un tamaño fijo. Un “bulkhead” sin límite es solo un pool compartido con pasos extra.
2. Fail fast al saturarse — cuando un pool está lleno, el llamador es rechazado de inmediato (BulkheadFullError, HTTP 429 o una tarea descartada). Bloquear dentro de un bulkhead devuelve la contención al llamador, que es justo lo que querías evitar.
3. Dominios de fallo independientes — un bulkhead saturado o roto no puede propagarse. El radio de impacto de una dependencia defectuosa queda acotado a una partición.

Elegir el tipo de bulkhead

Tipo de bulkhead	Aísla	Costo	Ideal para
Thread pool	Hilos de ejecución	Medio (hilos dedicados)	I/O bloqueante, llamadas síncronas
Semáforo	Permisos concurrentes	Bajo (sin hilos extra)	Código async, frameworks no bloqueantes
Pool de conexiones	Conexiones a BD/externas	Medio	Queries críticas vs no críticas
Cuota de recursos (K8s)	CPU/memoria/pods	Bajo (lo aplica la plataforma)	Aislamiento a nivel namespace

Un detalle que se pasa por alto: un semáforo limita la concurrencia, no los hilos. En código async eso suele ser lo que quieres: un solo hilo de event-loop puede sostener miles de permisos. En código bloqueante, un semáforo sin pool dedicado solo limita cuántos hilos pueden hacer cola, así que combínalo con un thread pool acotado o los hilos se amontonarán igual.

Dimensionar un bulkhead

Dimensiona desde la capacidad del servicio aguas abajo, no desde tu tráfico. Si la API de pagos maneja 50 requests concurrentes, dale a su bulkhead 10-20 slots: suficiente para usarla por completo pero dejando margen para reintentos y para otros consumidores de la dependencia. Para cargas no críticas, 3-5 suele estar bien: un bulkhead casi tan grande como el pool compartido que reemplazó no está aislando nada.

Vigila la tasa de rechazos, no solo la utilización. Un bulkhead al 60% de uso que no rechaza nada está sano; uno al 95% rechazando constantemente significa que el pool es chico o la dependencia se está degradando — ambos merecen una alerta, y ambos son invisibles si solo graficas la latencia promedio.

Variantes

Bulkhead por tenant

# bulkhead/per_tenant.py — Aislar recursos por tenant
import threading
from collections import defaultdict
from contextlib import contextmanager

class PerTenantBulkhead:
    """Cada tenant tiene su propio límite de concurrencia.
    Un tenant no puede agotar los recursos de todos los demás."""

    def __init__(self, max_per_tenant=10):
        self._max_per_tenant = max_per_tenant
        self._semaphores = defaultdict(lambda: threading.Semaphore(max_per_tenant))
        self._active = defaultdict(int)
        self._lock = threading.Lock()

    @contextmanager
    def acquire(self, tenant_id):
        sem = self._semaphores[tenant_id]
        if not sem.acquire(timeout=5):
            # from bulkhead.thread_pool import BulkheadFullError
            raise BulkheadFullError(
                f"El tenant {tenant_id} excedió {self._max_per_tenant} requests concurrentes"
            )
        with self._lock:
            self._active[tenant_id] += 1
        try:
            yield
        finally:
            with self._lock:
                self._active[tenant_id] -= 1
            sem.release()

    def get_stats(self):
        return dict(self._active)

Bulkhead híbrido + circuit breaker

# bulkhead/hybrid.py — Bulkhead con circuit breaker
import time
import threading

class BulkheadWithCircuitBreaker:
    """Combina bulkhead (límite de concurrencia) con circuit breaker
    (límite de fallos). El bulkhead evita el agotamiento de recursos;
    el circuit breaker evita llamar a un servicio que está fallando."""

    def __init__(self, max_concurrent, failure_threshold=5, recovery_timeout=30):
        self._semaphore = threading.Semaphore(max_concurrent)
        self._failure_count = 0
        self._failure_threshold = failure_threshold
        self._recovery_timeout = recovery_timeout
        self._circuit_open = False
        self._last_failure = 0
        self._lock = threading.Lock()

    def call(self, fn, *args, **kwargs):
        # Revisa el circuit breaker
        with self._lock:
            if self._circuit_open:
                if time.time() - self._last_failure > self._recovery_timeout:
                    self._circuit_open = False
                    self._failure_count = 0
                else:
                    # from bulkhead.exceptions import CircuitBreakerOpenError
                    raise CircuitBreakerOpenError("Circuit is open")

        # Adquiere el permiso del bulkhead
        if not self._semaphore.acquire(timeout=10):
            # from bulkhead.thread_pool import BulkheadFullError
            raise BulkheadFullError("Bulkhead is full")

        try:
            result = fn(*args, **kwargs)
            with self._lock:
                self._failure_count = 0
                self._circuit_open = False
            return result
        except Exception:
            with self._lock:
                self._failure_count += 1
                self._last_failure = time.time()
                if self._failure_count >= self._failure_threshold:
                    self._circuit_open = True
            raise
        finally:
            self._semaphore.release()


class CircuitBreakerOpenError(Exception):
    pass

Buenas Prácticas

• Dimensiona los bulkheads según el SLA: los servicios críticos reciben más recursos, los no críticos menos
• Monitorea la utilización de los bulkheads: alerta cuando el uso supere el 80%
• Usa pools separados para rutas críticas y no críticas: nunca las compartas
• Define la capacidad de la cola deliberadamente: las colas sin límite ocultan la contrapresión
• Combina con circuit breakers: los bulkheads limitan la concurrencia, los circuit breakers dejan de llamar a servicios que fallan. Consulta el patrón Circuit Breaker para la mitad de detección de fallos
• Usa timeouts dentro de los bulkheads: un hilo atascado en una llamada lenta sigue ocupando un slot
• Prueba el comportamiento del bulkhead: inyecta latencia y verifica que el aislamiento funciona

Errores Comunes

• Thread pool compartido para todos los servicios: anula el propósito. Un servicio lento deja sin recursos a todos los demás.
• Sin límite de capacidad de cola: las tareas se acumulan en una cola sin tope, consumiendo memoria.
• Bulkhead demasiado grande: si el pool es casi tan grande como el total, no hay aislamiento.
• Sin timeout al adquirir el bulkhead: los llamadores se bloquean indefinidamente esperando un permiso.
• No monitorear los rechazos del bulkhead: fallos silenciosos cuando el bulkhead está lleno. Alerta sobre la tasa de rechazos.

Lectura Adicional

• Bulkhead pattern — Microsoft Azure Architecture Center — la documentación del patrón con contexto, problemas y consideraciones.
• Documentación de Bulkhead de Resilience4j — las implementaciones SemaphoreBulkhead y ThreadPoolBulkhead usadas en el ejemplo Java.
• REL10-BP03 Use bulkhead architectures — AWS Well-Architected — el enfoque de arquitectura basada en celdas a nivel de infraestructura.
• Release It! de Michael Nygard — el libro que popularizó el patrón; el capítulo de patrones de estabilidad cubre bulkheads y circuit breakers.