Autoescalado de workers para resolver CAPTCHA

El número ideal de workers para resolver CAPTCHA no es un valor fijo: cambia cada minuto según cuántas tareas tengas en cola. El autoescalado resuelve eso atando el tamaño del pool a la demanda real —profundidad de la cola, saldo disponible y latencia—, de modo que no pagues infraestructura ociosa en las horas valle ni te quedes corto durante un pico.

Para una agencia que factura en pesos o euros pero paga la API de CaptchaAI en USD, ese ajuste automático es la diferencia entre un costo mensual predecible y una factura de servidores inflada. En esta guía verás tres patrones listos para producción —hilos, procesos y Kubernetes HPA— con control de saldo incluido.

Cuándo escalar y cuándo reducir

Antes de tocar código, define las señales que disparan cada decisión. Estas cinco son las que mejor funcionan en un pool de resolución de CAPTCHA:

La asimetría es la clave: amplía rápido ante un pico, pero reduce con calma cuando la cola se vacía. Así el pool no entra y sale de estados con cada ráfaga corta.

Qué estrategia de escalado elegir

No todos los proyectos necesitan Kubernetes. Ubica dónde está tu cuello de botella y quédate con el patrón más simple que lo resuelva:

La decisión se reduce a tres casos:

• Grupo de hilos: tu carga es puro I/O —llamadas a la API de CaptchaAI— y quieres la mínima latencia y complejidad.
• Grupo de procesos: además de resolver, haces preprocesamiento de imágenes u otro trabajo que satura la CPU.
• Kubernetes HPA o KEDA: ya operas en la nube y prefieres escalar pods con la infraestructura que ya tienes.

Autoescalado basado en hilos

Para cargas I/O-bound —y las llamadas a la API de CaptchaAI lo son, porque el trabajo pesado ocurre en el servidor de resolución— los hilos son la opción más simple y de menor latencia. Este pool arranca con un mínimo de hilos y ajusta su tamaño en un bucle en segundo plano:

import os
import time
import threading
import requests
import json
import redis


class AutoScalingPool:
    """Dynamically scale CaptchaAI worker threads."""

    def __init__(self, api_key, redis_url="redis://localhost:6379"):
        self.api_key = api_key
        self.redis = redis.from_url(redis_url)
        self.base = "https://ocr.captchaai.com"
        self.queue_key = "captcha:tasks"
        self.results_key = "captcha:results"

        self.min_workers = 2
        self.max_workers = 20
        self.workers = []
        self.active_count = 0
        self.lock = threading.Lock()
        self.running = True

    def start(self):
        """Start the pool with minimum workers."""
        for _ in range(self.min_workers):
            self._add_worker()

        # Start scaler in background
        scaler = threading.Thread(target=self._scaling_loop, daemon=True)
        scaler.start()
        print(f"Pool started with {self.min_workers} workers")

    def _add_worker(self):
        """Add a worker thread."""
        if len(self.workers) >= self.max_workers:
            return
        t = threading.Thread(target=self._worker_loop, daemon=True)
        t.start()
        self.workers.append(t)

    def _remove_worker(self):
        """Signal one worker to stop (lazy removal)."""
        if len(self.workers) <= self.min_workers:
            return
        self.workers.pop()  # Thread will exit on next idle cycle

    def _worker_loop(self):
        """Worker loop: fetch and process tasks."""
        while self.running and threading.current_thread() in self.workers:
            result = self.redis.blpop(self.queue_key, timeout=10)
            if result is None:
                continue

            _, raw = result
            task = json.loads(raw)
            task_id = task["id"]

            with self.lock:
                self.active_count += 1

            try:
                token = self._solve(task["method"], task["params"])
                self.redis.hset(self.results_key, task_id, json.dumps({
                    "status": "success", "token": token,
                }))
            except Exception as e:
                self.redis.hset(self.results_key, task_id, json.dumps({
                    "status": "error", "error": str(e),
                }))
            finally:
                with self.lock:
                    self.active_count -= 1

    def _scaling_loop(self):
        """Periodically adjust worker count."""
        while self.running:
            time.sleep(10)

            queue_depth = self.redis.llen(self.queue_key)
            current = len(self.workers)
            utilization = (
                self.active_count / current * 100 if current > 0 else 0
            )

            # Scale up: queue growing and workers busy
            if queue_depth > 20 and utilization > 70:
                new_count = min(current + 2, self.max_workers)
                while len(self.workers) < new_count:
                    self._add_worker()
                print(f"Scaled up: {current} → {len(self.workers)} workers")

            # Scale down: queue empty and workers idle
            elif queue_depth < 5 and utilization < 20:
                target = max(current - 1, self.min_workers)
                while len(self.workers) > target:
                    self._remove_worker()
                if len(self.workers) < current:
                    print(f"Scaled down: {current} → {len(self.workers)} workers")

    def _solve(self, method, params, timeout=120):
        data = {"key": self.api_key, "method": method, "json": 1}
        data.update(params)

        resp = requests.post(
            f"{self.base}/in.php", data=data, timeout=30,
        )
        result = resp.json()

        if result.get("status") != 1:
            raise RuntimeError(result.get("request"))

        captcha_id = result["request"]
        start = time.time()

        while time.time() - start < timeout:
            time.sleep(5)
            resp = requests.get(f"{self.base}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": captcha_id,
                "json": 1,
            }, timeout=15)
            data = resp.json()
            if data["request"] != "CAPCHA_NOT_READY":
                if data.get("status") == 1:
                    return data["request"]
                raise RuntimeError(data["request"])

        raise TimeoutError("Solve timeout")

    def stats(self):
        return {
            "workers": len(self.workers),
            "active": self.active_count,
            "queue": self.redis.llen(self.queue_key),
        }


# Usage
pool = AutoScalingPool(os.environ["CAPTCHAAI_KEY"])
pool.start()

# Monitor
while True:
    print(pool.stats())
    time.sleep(30)

El pool crece de dos en dos cuando la cola supera las 20 tareas y los hilos están ocupados por encima del 70%, y retrocede de uno en uno cuando la cola se vacía. Antes de llevarlo a producción, ajusta tres valores a tu carga:

• min_workers: los hilos que quieres siempre vivos para absorber el arranque sin latencia.
• max_workers: el techo de concurrencia; no lo subas por encima de los threads de tu plan (ver más abajo), porque no ganarías paralelismo real.
• Los umbrales de cola (> 20) y de utilización (> 70%): bájalos si tu tráfico llega en ráfagas muy cortas.

Autoescalado basado en procesos

Si además de resolver haces preprocesamiento de imágenes o cálculos que saturan la CPU, los hilos de Python chocan con el GIL y dejan de sumar. Ahí conviene escalar procesos: aíslan la CPU aunque cuesten más memoria y un arranque más lento.

import multiprocessing
import time
import redis
import os


class ProcessScaler:
    """Scale worker processes based on queue depth."""

    def __init__(self, worker_fn, redis_url="redis://localhost:6379"):
        self.worker_fn = worker_fn
        self.redis = redis.from_url(redis_url)
        self.processes = []
        self.min_workers = 2
        self.max_workers = 16

    def run(self, check_interval=15):
        """Run the scaler loop."""
        # Start minimum workers
        for _ in range(self.min_workers):
            self._spawn()

        while True:
            time.sleep(check_interval)
            self._cleanup_dead()

            queue_depth = self.redis.llen("captcha:tasks")
            current = len(self.processes)

            # Scale up
            if queue_depth > current * 5 and current < self.max_workers:
                to_add = min(
                    max(1, queue_depth // 10),
                    self.max_workers - current,
                )
                for _ in range(to_add):
                    self._spawn()
                print(f"Scaled up to {len(self.processes)} workers")

            # Scale down
            elif queue_depth < 3 and current > self.min_workers:
                to_remove = min(2, current - self.min_workers)
                for _ in range(to_remove):
                    p = self.processes.pop()
                    p.terminate()
                print(f"Scaled down to {len(self.processes)} workers")

    def _spawn(self):
        p = multiprocessing.Process(target=self.worker_fn)
        p.start()
        self.processes.append(p)

    def _cleanup_dead(self):
        self.processes = [p for p in self.processes if p.is_alive()]
        # Ensure minimum
        while len(self.processes) < self.min_workers:
            self._spawn()

_cleanup_dead() recoge los procesos caídos en cada vuelta y repone el mínimo, para que un worker que muera no deje la cola desatendida. Frente al pool de hilos, hay dos diferencias que tener presentes:

• Cada proceso arranca su propio intérprete de Python, así que consume más memoria y tarda más en estar listo.
• A cambio, un proceso que falle o se cuelgue no arrastra a los demás: el aislamiento es real.

Frenar el escalado según el saldo

Escalar sin mirar el saldo es la forma más rápida de terminar con muchos workers activos y sin fondos para resolver nada. CaptchaAI expone el saldo con la acción getbalance, así que consúltalo antes de sumar capacidad:

def check_balance(api_key, min_balance=2.0):
    """Check if balance is sufficient for scaling."""
    resp = requests.get("https://ocr.captchaai.com/res.php", params={
        "key": api_key,
        "action": "getbalance",
        "json": 1,
    }, timeout=15)
    balance = float(resp.json()["request"])

    if balance < min_balance:
        print(f"Balance ${balance:.2f} below ${min_balance} — halting scale-up")
        return False
    return True

Conéctalo al ciclo de escalado para que un saldo bajo pause el crecimiento en lugar de romper el pool:

# In _scaling_loop:
if queue_depth > 20 and utilization > 70:
    if check_balance(self.api_key, min_balance=2.0):
        # Scale up
        ...
    else:
        print("Scaling paused — low balance")

Recuerda que CaptchaAI cobra por thread concurrente contratado, no por resolución: el saldo se consume con el uso, y esta comprobación evita que sigas levantando workers cuando ya no hay margen.

Preguntas frecuentes

¿El autoescalado reduce mi gasto en CaptchaAI?

No directamente. CaptchaAI cobra por thread concurrente contratado, no por resolución, así que tu factura depende del plan —por ejemplo, BASIC ($15/mes, 5 threads) o ADVANCE ($90/mes, 50 threads)—, no de cuántos workers levantes. Lo que el autoescalado sí reduce es el gasto en tu propia infraestructura (CPU, memoria, instancias) durante las horas de baja carga.

¿Cuántos workers puedo ejecutar según mi plan?

Tantos como quieras, pero solo tendrás tantas resoluciones en paralelo como threads incluya tu plan. Si contratas STANDARD ($30/mes, 15 threads), levantar 40 hilos de worker no acelera nada: 15 estarán resolviendo y el resto esperará turno. Ajusta max_workers a los threads de tu plan para no desperdiciar memoria.

¿Cómo evito que el pool oscile sin parar?

Escala rápido y reduce despacio. Comprueba las señales cada 10 a 15 segundos para crecer, pero exige entre 30 y 60 segundos de carga baja sostenida antes de reducir. Esa asimetría evita que el pool entre y salga de estados una y otra vez ante picos cortos.

¿Conviene usar KEDA o el HPA de Kubernetes?

Si ya despliegas en Kubernetes, sí: el HPA escala pods por CPU o métricas personalizadas, y KEDA añade escalado directo por la profundidad de la cola en Redis. Para un solo servidor, el grupo de hilos o procesos de esta guía te da el mismo resultado con mucha menos complejidad.

Problemas frecuentes y cómo resolverlos

Cuando el pool se comporta mal, casi siempre es una de estas causas:

Guías relacionadas

• Colas de trabajos en Kubernetes para resolver CAPTCHA a escala
• Monitoreo de tasas de resolución con Prometheus y Grafana

Escala con cabeza: consigue tu API key de CaptchaAI y ajusta tus workers a la demanda real desde el primer día.