Worker Pemecahan CAPTCHA dengan Auto-Scaling

Worker pool statis nyaris selalu salah ukuran: terlalu besar saat sepi sehingga thread menganggur, atau terlalu kecil saat lonjakan sehingga antrean menumpuk dan SLA meleset. Auto-scaling menyesuaikan jumlah worker real-time berdasarkan antrean, utilisasi, dan saldo — bukan tebakan kasar di awal deployment. Tiga pola berikut teruji di produksi: thread pool untuk I/O-bound, process pool untuk preprocessing berat CPU, dan opsi cloud-native lewat Kubernetes/KEDA.

Konteks lokal: agensi price-monitoring dan developer scraping lepas di Indonesia sering menjalankan worker ini di AWS ap-southeast-3 (Jakarta) atau GCP asia-southeast2 agar round-trip ke CaptchaAI tetap singkat. CaptchaAI menagih per thread — BASIC ($15/bulan, 5 thread) sampai ENTERPRISE ($300/bulan, 200 thread) — jadi auto-scaling presisi langsung memengaruhi biaya bulanan, bukan cuma kecepatan.

Checklist Kesiapan Sebelum Auto-Scaling

Sebelum menyalakan scaler, pastikan lima hal ini sudah beres — kebanyakan masalah produksi bukan soal algoritma scaling, tapi fondasi yang belum siap:

1. Redis (atau broker antrean lain) sudah berjalan dan bisa diakses semua worker, bukan cuma dari mesin development.
2. min_workers dan max_workers sudah ditentukan dari beban puncak historis, bukan angka tebakan.
3. Notifikasi saldo sudah terpasang — scaler yang jalan tanpa pengecekan saldo bisa menghabiskan thread dalam hitungan menit saat lonjakan.
4. Dashboard monitoring (Datadog atau Grafana) sudah menampilkan kedalaman antrean, worker aktif, dan latensi P95 secara real-time.
5. Logika scaling sudah diuji di staging dengan simulasi lonjakan trafik sebelum dipasang di produksi.

Pilih Dulu: Thread Pool, Process Pool, atau Kubernetes?

Sebelum menyelami kode, tentukan pola mana yang cocok dengan beban kerja Anda — pilihan ini menentukan seberapa rumit setup scaling yang perlu dibangun:

Sinyal Auto-Scaling: Kapan Naik, Kapan Turun

Auto-Scaling Worker Berbasis Thread Pool

Pola scaling thread worker dalam satu proses — tepat karena panggilan CaptchaAI adalah operasi I/O-bound:

• Thread cukup menunggu respons dari in.php/res.php, bukan menunggu CPU — jadi tidak ada alasan memakai proses yang lebih berat untuk kasus ini.
• Startup thread jauh lebih murah dibanding spawn proses baru, jadi scale-up dan scale-down bisa berjalan cepat tanpa jeda berarti.
• Cocok untuk mayoritas worker CaptchaAI: yang hanya submit task dan polling hasil, tanpa preprocessing berat sebelum request.

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)

Auto-Scaling Worker Berbasis Process Pool

Kalau worker melakukan preprocessing gambar atau komputasi berat sebelum memanggil CaptchaAI, isolasi lewat proses — bukan thread — mencegah satu tugas memblokir thread lain (GIL Python):

• Ada tahap preprocessing gambar (resize, denoise, crop) sebelum task dikirim ke CaptchaAI.
• Workload benar-benar CPU-bound, jadi GIL Python akan membuat thread pool terasa lambat meski jumlah worker ditambah.
• Anda butuh isolasi kegagalan — satu proses preprocessing yang crash tidak boleh menjatuhkan worker lain di pool yang sama.

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

Scaling yang Memperhitungkan Saldo API

Auto-scaling agresif tanpa pengecekan saldo bisa menghabiskan kuota thread dalam hitungan menit. Validasi saldo sebelum menambah worker baru:

• Paket kecil seperti BASIC ($15/bulan, 5 thread) bisa kehabisan saldo dalam hitungan menit kalau scaler naik terus tanpa jeda saat lonjakan tugas.
• Volume tinggi di ADVANCE ($90/bulan, 50 thread) ke atas tetap butuh ambang saldo — makin besar paketnya, makin cepat pula saldo terkuras kalau scaling tidak dibatasi.
• Tentukan min_balance dari burn rate historis Anda, bukan angka acak — cek berapa saldo yang biasanya habis per jam saat trafik puncak.

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

Pasang pengecekan ini langsung di loop scaling:

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

Kapan Auto-Scaling Tidak Diperlukan (atau Justru Berisiko)

Auto-scaling bukan default yang wajib untuk semua tim. Ada situasi di mana worker pool statis tetap pilihan yang lebih masuk akal:

• Volume stabil dan terprediksi — misalnya cron job harian dengan beban yang selalu mirip setiap hari — worker pool statis lebih sederhana dan lebih gampang di-debug daripada scaler yang jarang benar-benar dipakai.
• Tim kecil dengan volume tugas yang sangat rendah biasanya menanggung overhead mengelola scaler yang lebih besar daripada penghematan biayanya.
• max_workers yang terlalu longgar tanpa pengecekan saldo bisa memicu lonjakan permintaan mendadak ke CaptchaAI dan menguras saldo lebih cepat dari perkiraan — auto-scaling yang agresif tanpa batas jelas adalah risiko, bukan jaminan efisiensi.

Troubleshooting Auto-Scaling Worker

Sebagian besar masalah auto-scaling berasal dari ambang batas yang terlalu sensitif atau saldo yang tidak dipantau, bukan dari bug di kode scaler itu sendiri. Pola berikut paling sering muncul di produksi:

FAQ Auto-Scaling Worker CaptchaAI

Berapa rasio worker terhadap antrean yang paling pas?

Rule of thumb: 1 worker per 5–10 tugas mengantre. Satu worker memproses sekitar 3–6 CAPTCHA per menit — reCAPTCHA v3 dan Turnstile lebih cepat daripada image atau grid CAPTCHA, jadi sesuaikan dengan campuran tipe di antrean Anda.

Sebaiknya pakai thread atau process untuk scaling worker?

Pakai thread kalau worker murni memanggil API CaptchaAI (I/O-bound). Pindah ke process untuk preprocessing gambar atau komputasi berat yang berjalan bersamaan — GIL Python membuat thread kurang efektif untuk CPU-bound.

Berapa thread CaptchaAI yang saya butuhkan sebelum mulai auto-scaling?

BASIC ($15/bulan, 5 thread) cukup untuk uji coba; tim produksi biasanya naik ke ADVANCE ($90/bulan, 50 thread). Semua paket unlimited solve per thread — auto-scaling di sini soal alokasi thread, bukan biaya per solve.

Apakah auto-scaling tetap perlu untuk tim kecil atau proyek scraping lepas?

Perlu, dengan skala lebih kecil. min_workers=2 dan max_workers 8–10 wajar untuk agensi price-monitoring atau freelancer bervolume rendah — intinya sama: jangan bayar thread menganggur, jangan biarkan antrean menumpuk saat lonjakan. Pantau antrean, worker aktif, dan latensi P95 lewat Datadog atau Grafana.

Apakah auto-scaling memengaruhi tagihan bulanan CaptchaAI?

Tidak langsung. Semua paket CaptchaAI unlimited solve per thread, jadi tagihan tetap flat sesuai paket yang dipilih — BASIC ($15/bulan) sampai ENTERPRISE ($300/bulan). Auto-scaling menentukan berapa banyak thread yang Anda BUTUHKAN dari alokasi paket tersebut, bukan biaya per solve. Kalau worker sering mentok di max_workers, itu sinyal untuk naik tier, bukan tanda tagihan tiba-tiba membengkak.

Panduan Terkait

• Bangun antrean penyelesaian CAPTCHA dengan Python
• Pantau metrik CaptchaAI dengan Datadog

Scaling cerdas, bukan tebakan — ambil kunci API CaptchaAI Anda dan mulai hari ini.