Mettre à l'échelle automatiquement des workers de résolution CAPTCHA

Combien de workers faut-il pour résoudre vos CAPTCHA ? La bonne réponse n'est jamais un chiffre fixe : elle change avec le trafic, l'heure et le type de CAPTCHA. Dimensionner un pool statique revient soit à payer une capacité qui dort, soit à saturer dès le premier pic. L'auto-scaling supprime ce compromis en ajustant le nombre de workers à la demande réelle, mesurée en continu.

Le vrai enjeu n'est pas d'ajouter des workers, mais de le faire au bon moment et sur les bons signaux, sans faire osciller le pool ni vider le solde du compte. Cet article couvre les signaux à surveiller, les stratégies possibles et le code Python pour les implémenter avec CaptchaAI.

Les signaux qui déclenchent le scaling

Un seul indicateur ne suffit pas : c'est leur combinaison qui évite les décisions à contretemps. Une file qui gonfle avec des workers déjà saturés justifie une montée ; une file vide avec des workers oisifs appelle une descente. Traitez ces seuils comme des points de départ à ajuster sur vos propres mesures.

Le solde n'est jamais un signal de montée : il ne fait que bloquer un scale-up quand les fonds manquent.

Choisir une stratégie avant d'écrire du code

Choisissez d'abord le modèle d'exécution adapté à votre charge. La résolution via CaptchaAI est surtout orientée I/O : un pool de threads suffit dans la plupart des cas. Vous ne passez aux processus ou à l'orchestrateur que pour du travail CPU ou un déploiement cloud natif.

Commencez par la ligne la plus simple qui couvre votre besoin, puis remontez ce tableau seulement si un plafond concret vous y oblige :

• charge purement I/O (appels API et polling) → un pool de threads ;
• prétraitement CPU en plus de la résolution → un pool de processus ;
• déploiement cloud natif → l'autoscaler HPA ou KEDA.

Scaler des threads dans un seul processus

Quand le worker passe surtout son temps à attendre des résultats d'API, un pool de threads est la stratégie la plus économe. Le pool démarre au minimum, puis une boucle ajuste le nombre de threads toutes les dix secondes selon la profondeur de file et l'utilisation.

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)

Trois paramètres pilotent le comportement :

• min_workers maintient une réactivité minimale, même à froid ;
• max_workers plafonne le coût et protège l'API en aval ;
• la boucle observe toutes les 10 s, un rythme adapté à une charge I/O.

Scaler des processus pour les tâches CPU

Dès que vous ajoutez du prétraitement d'image ou d'autres opérations gourmandes en CPU, isoler les workers dans des processus distincts devient plus propre : chaque processus a son propre GIL et un plantage n'emporte pas les autres.

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()

Garder un œil sur le solde

Un bon auto-scaler ne regarde pas que la file d'attente. Si le solde descend trop bas, monter en charge ne fait qu'accélérer l'épuisement du compte. Une vérification légère du solde, branchée sur la boucle de scaling, sert de garde-fou final.

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

Fixez un min_balance strictement positif : viser un plancher au-dessus de zéro vous laisse le temps de recharger avant l'arrêt complet du scale-up.

Branchez la vérification directement dans la décision de montée :

# 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")

Un pic de trafic, étape par étape

Prenez un worker déployé sur OVHcloud ou Scaleway, en région eu-west-3 (Paris), qui traite un flux régulier de reCAPTCHA v2. Un pic déclenché par une campagne du soir se déroule ainsi :

1. À 20 h, le volume double en quelques minutes : la file passe de 8 à 60.
2. L'utilisation des workers dépasse 90 % et la latence P95 franchit le seuil.
3. Les trois signaux de montée s'alignent ; le scaler ajoute des workers par paliers de deux.
4. Une fois la file résorbée, le scale-down ramène le pool à son minimum.

Un point mérite attention côté facturation. CaptchaAI facture au thread concurrent, pas à la résolution : un plan BASIC ($15/mois, 5 threads) autorise cinq CAPTCHA en vol simultanément, avec des résolutions illimitées par thread. Lancer vingt workers quand le plan plafonne à cinq threads ne fait qu'allonger la file côté API. Calez max_workers sur la capacité de votre plan — STANDARD 15 threads, ADVANCE 50 — et laissez le solde jouer le dernier garde-fou. Côté RGPD, ne journalisez que le nécessaire : un task_id et un statut suffisent.

Dépannage

FAQ

L'auto-scaling augmente-t-il le nombre de threads facturés par CaptchaAI ?

Non. Le forfait est mensuel et se paie par threads concurrents ; ajouter des workers au-delà de ce plafond n'augmente pas la facture, cela allonge l'attente côté API. Alignez max_workers sur les threads de votre plan.

Comment éviter que le scale-up ne vide mon solde ?

Ajoutez une vérification du solde avant chaque montée, avec un seuil plancher. Sous ce seuil, le scaler suspend la montée, ce qui vous laisse le temps de recharger sans interrompre le service.

Threads ou processus pour mes workers ?

Des threads si la charge est purement I/O, comme la résolution via l'API CaptchaAI. Passez aux processus dès que vous ajoutez du prétraitement d'image ou d'autres calculs CPU.

Kubernetes HPA ou KEDA pour un scaling piloté par la file ?

HPA réagit surtout au CPU et à la mémoire, ce qui colle mal à une charge I/O. Pour scaler sur la profondeur d'une file Redis, KEDA est plus direct : il expose la longueur de file comme métrique.

Guides connexes

• Files d'attente de tâches sur Kubernetes
• Superviser vos taux de résolution avec Prometheus et Grafana

Pour absorber les pics sans surpayer au repos, créez votre clé CaptchaAI et pilotez vos workers avec des signaux concrets plutôt qu'un nombre fixe.