Patron de Programador-Agente-Supervisor

Resumen

El Patron de Programador-Agente-Supervisor estructura el procesamiento de trabajos distribuidos en tres roles distintos: un Programador que decide que trabajo hacer y cuando, Agentes que ejecutan el trabajo real, y un Supervisor que monitorea agentes, maneja fallas y gestiona el ciclo de vida del sistema.

Esta separacion de responsabilidades hace al sistema resiliente ante caidas de agentes, particiones de red y workers no responsivos. El supervisor detecta agentes fallidos, los reinicia, redistribuye su trabajo y mantiene la programacion funcionando incluso cuando componentes individuales fallan. Es la base de muchos sistemas de orquestacion, desde los arboles de supervision OTP de Erlang hasta los controladores de Kubernetes.

Cuando Usar

Para alternativas, ver Patron de Back-Pressure.

• Trabajos de larga duracion o de fondo que deben sobrevivir fallas de maquinas individuales — he visto pipelines de procesamiento batch donde una sola caida de worker significaba reiniciar todo el trabajo manualmente
• Ejecucion de tareas distribuidas donde los workers pueden fallar o volverse inalcanzables — piensa en instancias spot o contenedores que el orquestador mata
• Sistemas que requieren recuperacion automatica, reintentos y aislamiento de fallas — si estas reiniciando trabajos fallidos manualmente a las 3 AM, necesitas este patron
• Flujos de trabajo donde cada paso se supervisa independientemente — cada paso puede fallar y reintentarse sin matar todo el workflow
• Entornos donde los ejecutores de trabajos (agentes) corren en infraestructura heterogenea o efimera — instancias cloud que van y vienen

Cuando Evitar

• Trabajos cron simples en un solo servidor donde gestores de procesos (systemd, PM2) son suficientes — no construyas un supervisor cuando una politica de reinicio hace el trabajo
• Manejo de solicitudes HTTP sin estado donde las fallas se muestran inmediatamente al llamador — la supervision agrega latencia sin beneficio
• Sistemas donde el overhead de supervision excede el costo de fallas ocasionales — si un trabajo fallido cuesta $5 y la supervision cuesta $500, saltatelo
• Tareas de muy corta duracion donde la eleccion de supervisor y verificaciones de heartbeat agregan latencia inaceptable — el overhead de supervision domina el trabajo real

Si necesitas manejar trabajos fallidos sin supervision completa, ver el Runbook de Dead Letter Queue para un enfoque mas ligero de manejo de trabajos fallidos.

Solucion

Python (Supervisor Personalizado con Agentes)

import asyncio
import time
import uuid
from dataclasses import dataclass, field
from typing import Dict, Optional, Callable
from enum import Enum

class JobStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class Job:
    id: str
    task: Callable
    status: JobStatus = JobStatus.PENDING
    agent_id: Optional[str] = None
    retries: int = 0
    max_retries: int = 3

class Agent:
    def __init__(self, agent_id: str):
        self.id = agent_id
        self.current_job: Optional[Job] = None
        self.last_heartbeat = time.time()

    async def execute(self, job: Job) -> bool:
        self.current_job = job
        job.status = JobStatus.RUNNING
        job.agent_id = self.id
        try:
            result = await asyncio.wait_for(self._run_job(job), timeout=30.0)
            job.status = JobStatus.COMPLETED
            return True
        except Exception as e:
            job.status = JobStatus.FAILED
            return False
        finally:
            self.current_job = None
            self.last_heartbeat = time.time()

    async def _run_job(self, job: Job):
        await asyncio.sleep(0.5)
        if callable(job.task):
            return job.task()
        return job.task

    def is_healthy(self) -> bool:
        return time.time() - self.last_heartbeat < 15

class Supervisor:
    def __init__(self, check_interval: float = 10.0):
        self.agents: Dict[str, Agent] = {}
        self.jobs: Dict[str, Job] = {}
        self.check_interval = check_interval

    def add_agent(self, agent: Agent):
        self.agents[agent.id] = agent

    def submit_job(self, task: Callable) -> str:
        job = Job(id=str(uuid.uuid4()), task=task)
        self.jobs[job.id] = job
        return job.id

    async def start(self):
        asyncio.create_task(self._monitor_loop())

    async def _monitor_loop(self):
        while True:
            await asyncio.sleep(self.check_interval)
            await self._check_agents()
            await self._redistribute_work()

    async def _check_agents(self):
        failed = [aid for aid, a in self.agents.items() if not a.is_healthy()]
        for aid in failed:
            agent = self.agents[aid]
            if agent.current_job and agent.current_job.status == JobStatus.RUNNING:
                agent.current_job.status = JobStatus.FAILED
                agent.current_job.retries += 1
            del self.agents[aid]
            new_agent = Agent(f"agent-{uuid.uuid4().hex[:8]}")
            self.add_agent(new_agent)

    async def _redistribute_work(self):
        available = [a for a in self.agents.values() if a.current_job is None and a.is_healthy()]
        pending = [j for j in self.jobs.values() if j.status in (JobStatus.PENDING, JobStatus.FAILED) and j.retries < j.max_retries]
        for job in pending:
            if not available:
                break
            agent = available.pop(0)
            if job.status == JobStatus.FAILED:
                job.status = JobStatus.PENDING
            asyncio.create_task(self._execute_job(agent, job))

    async def _execute_job(self, agent: Agent, job: Job):
        success = await agent.execute(job)
        if not success and job.retries < job.max_retries:
            job.retries += 1
            job.status = JobStatus.PENDING

Java (Supervision Estilo Akka)

public class SchedulerAgentSupervisor {
    private final ScheduledExecutorService scheduler;
    private final ExecutorService workerPool;
    private final Map<String, Agent> agents = new ConcurrentHashMap<>();
    private final Map<String, Job> jobs = new ConcurrentHashMap<>();
    private final AtomicInteger agentCounter = new AtomicInteger(0);
    private volatile boolean running;

    public SchedulerAgentSupervisor(int workerPoolSize) {
        this.scheduler = Executors.newScheduledThreadPool(2);
        this.workerPool = Executors.newFixedThreadPool(workerPoolSize);
    }

    public void start() {
        running = true;
        scheduler.scheduleAtFixedRate(this::monitorAgents, 5, 5, TimeUnit.SECONDS);
        scheduler.scheduleAtFixedRate(this::schedulePendingJobs, 1, 1, TimeUnit.SECONDS);
    }

    public String spawnAgent() {
        String agentId = "agent-" + agentCounter.incrementAndGet();
        agents.put(agentId, new Agent(agentId));
        return agentId;
    }

    public String submitJob(Runnable task) {
        String jobId = "job-" + UUID.randomUUID().toString().substring(0, 8);
        jobs.put(jobId, new Job(jobId, task));
        return jobId;
    }

    private void monitorAgents() {
        List<String> failed = new ArrayList<>();
        for (Agent agent : agents.values()) {
            if (!agent.isHealthy()) {
                failed.add(agent.getId());
                if (agent.getCurrentJob() != null) {
                    agent.getCurrentJob().setStatus(JobStatus.FAILED);
                    agent.getCurrentJob().incrementRetries();
                }
            }
        }
        for (String id : failed) {
            agents.remove(id);
            spawnAgent();
        }
    }

    private void schedulePendingJobs() {
        List<Agent> available = agents.values().stream()
            .filter(a -> a.getCurrentJob() == null && a.isHealthy()).toList();
        List<Job> pending = jobs.values().stream()
            .filter(Job::canRetry).toList();
        int limit = Math.min(available.size(), pending.size());
        for (int i = 0; i < limit; i++) {
            dispatchJob(available.get(i), pending.get(i));
        }
    }

    private void dispatchJob(Agent agent, Job job) {
        job.setStatus(JobStatus.RUNNING);
        job.setAgentId(agent.getId());
        agent.setCurrentJob(job);
        workerPool.submit(() -> {
            try {
                job.getTask().run();
                job.setStatus(JobStatus.COMPLETED);
            } catch (Exception e) {
                job.setStatus(JobStatus.FAILED);
                job.incrementRetries();
            } finally {
                agent.setCurrentJob(null);
                agent.recordHeartbeat();
            }
        });
    }
}

JavaScript (Node.js con Clustering estilo PM2)

const cluster = require('cluster');
const os = require('os');

class JobSupervisor {
    constructor(options = {}) {
        this.workers = new Map();
        this.jobs = new Map();
        this.maxWorkers = options.maxWorkers || os.cpus().length;
        this.heartbeatTimeout = options.heartbeatTimeout || 10000;
        this.checkInterval = options.checkInterval || 5000;
    }

    start() {
        if (cluster.isPrimary) {
            this._startSupervisor();
        } else {
            this._startAgent();
        }
    }

    _startSupervisor() {
        for (let i = 0; i < this.maxWorkers; i++) {
            this._spawnWorker();
        }
        setInterval(() => this._monitorWorkers(), this.checkInterval);
        cluster.on('exit', (worker, code, signal) => {
            this.workers.delete(worker.id);
            this._spawnWorker();
        });
    }

    _spawnWorker() {
        const worker = cluster.fork();
        this.workers.set(worker.id, {
            worker, lastHeartbeat: Date.now(), currentJob: null
        });
    }

    _monitorWorkers() {
        const now = Date.now();
        for (const [id, info] of this.workers) {
            if (now - info.lastHeartbeat > this.heartbeatTimeout) {
                if (info.currentJob) this._requeueJob(info.currentJob);
                info.worker.kill('SIGTERM');
                this.workers.delete(id);
                this._spawnWorker();
            }
        }
    }

    submitJob(jobData) {
        const jobId = `job-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
        this.jobs.set(jobId, { id: jobId, data: jobData, status: 'pending', retries: 0 });
        for (const [id, info] of this.workers) {
            if (!info.currentJob) {
                this._dispatchJob(id, jobId);
                return jobId;
            }
        }
        return jobId;
    }

    _dispatchJob(workerId, jobId) {
        const info = this.workers.get(workerId);
        info.currentJob = jobId;
        const job = this.jobs.get(jobId);
        job.status = 'running'; job.workerId = workerId;
        info.worker.send({ type: 'execute', jobId, data: job.data });
    }

    _startAgent() {
        process.on('message', async (msg) => {
            if (msg.type === 'execute') {
                try {
                    const heartbeat = setInterval(() => {
                        process.send({ type: 'heartbeat', jobId: msg.jobId });
                    }, 3000);
                    const result = await this.executeJob(msg.data);
                    clearInterval(heartbeat);
                    process.send({ type: 'complete', jobId: msg.jobId, result });
                } catch (error) {
                    process.send({ type: 'failed', jobId: msg.jobId, error: error.message });
                }
            }
        });
    }

    async executeJob(data) {
        await new Promise(r => setTimeout(r, 1000));
        return { processed: true, data };
    }
}

Explicacion

El patron separa responsabilidades en tres capas:

• Programador: Decide que se ejecuta, cuando y donde. Mantiene la cola de trabajos, orden de prioridades y politicas de reintento — no ejecuta trabajo directamente.
• Agente: Ejecuta los trabajos asignados y reporta progreso. Los agentes son desechables — si uno falla, el supervisor lo reemplaza. Los agentes deben ser sin estado; todo el estado del trabajo vive en la tienda de trabajos del programador.
• Supervisor: Observa agentes via heartbeats, detecta fallas, reinicia agentes y redistribuye trabajos fallidos. Es la capa de resiliencia que hace al sistema auto-reparable.

La idea clave es que los agentes son ganado, no mascotas. El supervisor los trata como recursos efimeros que pueden crearse, destruirse y reemplazarse sin afectar el sistema. He visto equipos intentar preservar el estado del agente entre reinicios — nunca funciona bien. Diseña los agentes para ser desechables desde el dia uno.

Variantes

Variante	Mecanismo de Supervisor	Ideal Para
Erlang OTP	Arboles de supervision, reinicio one-for-one	Telecom, sistemas soft real-time
Kubernetes	ReplicaSets, health checks, reinicios rolling	Microservicios contenedorizados
AWS Step Functions	Maquina de estados con manejo de errores	Flujos de trabajo nativos en la nube
Celery con Flower	Monitoreo de workers, control remoto	Colas de tareas Python — ver la Guia de Message Queue para fundamentos de colas
Implementacion personalizada	Monitoreo directo de procesos	Sistemas embebidos, edge computing

Lo que funciona

• Mantener los agentes sin estado. He depurado sistemas donde los agentes cacheaban estado del trabajo en memoria — cada reinicio significaba perder progreso. Todo el estado del trabajo debe vivir en la tienda del programador para que cualquier agente pueda retomar donde otro lo dejo.
• Usar heartbeats con timeouts. Los agentes deben probar que estan vivos; heartbeats perdidos disparan reemplazo. He visto agentes que parecian saludables pero estaban atrapados en loops infinitos — los timeouts de heartbeat atrapan esto.
• Implementar backoff exponencial para reintentos. Un trabajo que falla inmediatamente probablemente falle de nuevo — no lo martillees con reintentos instantaneos. Agrega jitter para prevenir thundering herd en la recuperacion.
• Limitar el numero de reintentos. Un trabajo permanentemente roto no deberia reintentarse para siempre — muevelo a una dead letter queue despues del limite. He visto sistemas thrash por horas reintentando un trabajo que nunca iba a tener exito.
• Escalar los supervisores tambien. Un supervisor unico es un punto unico de falla — ejecutalo en pares con eleccion de lider o usa un sistema de consenso distribuido como etcd o ZooKeeper para el estado del supervisor.

Errores Comunes

• Agentes con estado. Un agente que almacena estado del trabajo localmente no puede reemplazarse sin perdida de datos — todo el estado del trabajo debe vivir en la tienda del programador, no en la memoria del agente.
• Heartbeats faltantes. Sin health checks, un agente caido aparece como “ejecutando” y su trabajo se estanca para siempre.
• Supervisor como punto unico de falla. Un supervisor gestionando cientos de agentes es en si mismo un riesgo de falla.
• Sin circuit breaker para trabajos fallidos. Un trabajo que falla instantaneamente en cada reintento agotara los limites de reintento rapidamente — combina este patron con el Patron de Circuit Breaker para detener fallas en cascada.
• Ignorar el tiempo de inicio del agente. Spawnear agentes de reemplazo toma tiempo — planifica para brechas temporales de capacidad.

Ejemplos del Mundo Real

Erlang/OTP

El framework OTP de Erlang fue pionero del patron de supervision. Los procesos se organizan en arboles de supervision donde los supervisores monitorean workers y aplican estrategias de reinicio (one-for-one, one-for-all, rest-for-one). Este diseno impulsa la infraestructura de mensajeria de WhatsApp, manejando millones de conexiones concurrentes con tolerancia a fallos.

Kubernetes

Los controladores de Kubernetes actuan como supervisores. Un controlador de Deployment monitorea Pods (agentes), detecta los no saludables via health checks, y crea reemplazos para mantener el conteo de replicas deseado. El programador (kube-scheduler) decide que nodo ejecuta cada Pod.

Apache Airflow

El modelo de executor de Airflow usa un programador que analiza DAGs y coloca tareas en workers (agentes). El programador monitorea el estado de las tareas en la base de datos de metadatos y reintenta tareas fallidas segun las politicas configuradas. Los executors Celery o Kubernetes distribuyen trabajo a traves de pools de agentes.