ELK Stack untuk Analisis Log Solve CAPTCHA

Sebuah agensi pemantauan harga di Jakarta yang berlangganan paket ADVANCE CaptchaAI ($90/bulan, 50 thread) pernah cerita: begitu volume task tembus 4.000–5.000 per hari, grep error worker-*.log yang tadinya sat-set berubah jadi puluhan detik — dan tetap tidak menjawab pertanyaan penting seperti "tipe CAPTCHA apa yang paling sering timeout minggu ini?".

ELK Stack (Elasticsearch, Logstash, Kibana) menjawab persis masalah ini: log JSON dari setiap worker dikirim ke satu tempat, bisa dicari dalam hitungan detik, dan divisualisasikan sebagai dashboard yang bisa dipantau tim on-call tanpa perlu SSH ke server satu per satu.

Sebelum Mulai: Checklist Cepat

Sebelum masuk ke konfigurasi, pastikan empat hal ini sudah siap:

• API key CaptchaAI aktif dan worker yang sudah bisa kirim task
• Worker menghasilkan log JSON per event (contoh formatnya di bawah)
• Docker/VM atau layanan terkelola untuk menjalankan Elasticsearch, Logstash, dan Kibana
• Akses tulis ke direktori log worker, misalnya /var/log/captcha-worker/

Arsitektur ELK untuk Log Worker CAPTCHA

Alurnya sederhana dan setiap komponen punya satu tugas: worker menulis log JSON, Filebeat menangkap dan mengirimkannya, Logstash mem-parsing dan memperkaya field, Elasticsearch menyimpannya sebagai index yang bisa di-query, dan Kibana menampilkan semuanya sebagai dashboard.

[CAPTCHA Workers] → JSON logs → [Filebeat] → [Logstash] → [Elasticsearch]
                                                                ↓
                                                           [Kibana]

Format Log JSON di Worker CAPTCHA

Sebelum ELK bisa membantu apa pun, worker Anda harus mengeluarkan log yang konsisten dan bisa di-parsing mesin — bukan baris teks bebas seperti print("gagal solve"). Setiap event minimal berisi timestamp, level, captcha_id, dan captcha_type, ditambah solve_time atau error_code kalau relevan.

Satu aturan yang paling sering dilanggar tim yang baru mulai: jangan pernah mencatat teks solusi/token CAPTCHA ke log. Token bersifat sekali pakai dan tidak punya nilai diagnostik, tapi menyimpannya menambah risiko keamanan dan biaya storage yang sama sekali tidak perlu — cukup catat metadata-nya saja.

Tips: untuk debug manual, tarik ulang hasil lewat captcha_id dan endpoint res.php — jangan mengandalkan token yang tersimpan di log.

Python — Output Log JSON

import os
import json
import time
import logging
import sys
import requests

API_KEY = os.environ["CAPTCHAAI_API_KEY"]


class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        # Add extra fields
        if hasattr(record, "captcha_id"):
            log_entry["captcha_id"] = record.captcha_id
        if hasattr(record, "captcha_type"):
            log_entry["captcha_type"] = record.captcha_type
        if hasattr(record, "solve_time"):
            log_entry["solve_time"] = record.solve_time
        if hasattr(record, "error_code"):
            log_entry["error_code"] = record.error_code
        if hasattr(record, "target_url"):
            log_entry["target_url"] = record.target_url
        if hasattr(record, "poll_count"):
            log_entry["poll_count"] = record.poll_count
        return json.dumps(log_entry)


# Configure logger
logger = logging.getLogger("captchaai")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)

session = requests.Session()


def solve_captcha(sitekey, pageurl, captcha_type="recaptcha_v2"):
    extra = {"captcha_type": captcha_type, "target_url": pageurl}

    # Submit
    resp = session.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": 1
    })
    data = resp.json()

    if data.get("status") != 1:
        logger.error("Submit failed", extra={
            **extra, "error_code": data.get("request")
        })
        return {"error": data.get("request")}

    captcha_id = data["request"]
    extra["captcha_id"] = captcha_id
    logger.info("Task submitted", extra=extra)

    # Poll
    start = time.time()
    poll_count = 0
    for _ in range(60):
        time.sleep(5)
        poll_count += 1
        result = session.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY, "action": "get", "id": captcha_id, "json": 1
        }).json()

        if result.get("status") == 1:
            elapsed = round(time.time() - start, 2)
            logger.info("Solve success", extra={
                **extra,
                "solve_time": elapsed,
                "poll_count": poll_count
            })
            return {"solution": result["request"]}

        if result.get("request") != "CAPCHA_NOT_READY":
            logger.error("Solve failed", extra={
                **extra,
                "error_code": result.get("request"),
                "poll_count": poll_count
            })
            return {"error": result.get("request")}

    logger.error("Solve timeout", extra={
        **extra,
        "error_code": "TIMEOUT",
        "poll_count": poll_count
    })
    return {"error": "TIMEOUT"}

Lima event di atas — submit gagal, task terkirim, solve sukses, solve gagal, dan timeout — cukup untuk melacak seluruh siklus hidup satu task lewat JSONFormatter yang sama.

Node.js — Format Log yang Sama di JavaScript

const axios = require("axios");

const API_KEY = process.env.CAPTCHAAI_API_KEY;

function log(level, message, fields = {}) {
  const entry = {
    timestamp: new Date().toISOString(),
    level,
    message,
    service: "captcha-worker",
    ...fields,
  };
  console.log(JSON.stringify(entry));
}

async function solveCaptcha(sitekey, pageurl, captchaType = "recaptcha_v2") {
  const fields = { captchaType, targetUrl: pageurl };

  const submitResp = await axios.post("https://ocr.captchaai.com/in.php", null, {
    params: {
      key: API_KEY, method: "userrecaptcha",
      googlekey: sitekey, pageurl, json: 1,
    },
  });

  if (submitResp.data.status !== 1) {
    log("error", "Submit failed", { ...fields, errorCode: submitResp.data.request });
    return { error: submitResp.data.request };
  }

  const captchaId = submitResp.data.request;
  fields.captchaId = captchaId;
  log("info", "Task submitted", fields);

  const startTime = Date.now();
  let pollCount = 0;

  for (let i = 0; i < 60; i++) {
    await new Promise((r) => setTimeout(r, 5000));
    pollCount++;

    const pollResp = await axios.get("https://ocr.captchaai.com/res.php", {
      params: { key: API_KEY, action: "get", id: captchaId, json: 1 },
    });

    if (pollResp.data.status === 1) {
      const solveTime = ((Date.now() - startTime) / 1000).toFixed(2);
      log("info", "Solve success", { ...fields, solveTime: parseFloat(solveTime), pollCount });
      return { solution: pollResp.data.request };
    }

    if (pollResp.data.request !== "CAPCHA_NOT_READY") {
      log("error", "Solve failed", { ...fields, errorCode: pollResp.data.request, pollCount });
      return { error: pollResp.data.request };
    }
  }

  log("error", "Solve timeout", { ...fields, errorCode: "TIMEOUT", pollCount });
  return { error: "TIMEOUT" };
}

module.exports = { solveCaptcha };

Field-nya sengaja dibuat sama dengan versi Python, supaya dashboard Kibana yang sama berlaku untuk worker di kedua bahasa.

Konfigurasi Filebeat untuk Kirim Log ke Logstash

Filebeat bertugas mengekor (tail) file log worker dan meneruskannya ke Logstash. Untuk sebagian besar setup single-server, konfigurasi minimal ini sudah cukup — tidak perlu agent tambahan atau plugin eksotis:

# filebeat.yml
filebeat.inputs:

  - type: log
    paths:

      - /var/log/captcha-worker/*.log
    json:
      keys_under_root: true
      add_error_key: true
      message_key: message

output.logstash:
  hosts: ["logstash:5044"]

keys_under_root: true membuat field JSON langsung jadi field top-level di Elasticsearch, bukan terkubur di dalam field message.

Pipeline Logstash: Parsing dan Field Turunan

Logstash menerima log dari Filebeat, mem-parsing JSON-nya, lalu menambahkan field yang tidak ada di log asli — misalnya solve_time_bucket untuk mengelompokkan solve cepat, sedang, dan lambat, supaya Anda tidak perlu menghitung ulang di setiap query Kibana.

Kalau worker dan cluster ELK Anda berjalan lintas region, latensi pengiriman log ikut terasa. Sebagian besar tim di Indonesia menjalankan cluster ELK di region yang sama dengan worker CAPTCHA mereka — AWS ap-southeast-1 (Singapura) atau GCP asia-southeast2 (Jakarta) — supaya jalur log tidak menambah beban pada jalur solve CAPTCHA itu sendiri.

# logstash-captcha.conf
input {
  beats {
    port => 5044
  }
}

filter {
  # Parse JSON logs
  json {
    source => "message"
    target => "captcha"
  }

  # Add computed fields
  if [captcha][solve_time] {
    mutate {
      add_field => {
        "solve_time_bucket" => "fast"
      }
    }
    if [captcha][solve_time] > 30 {
      mutate { update => { "solve_time_bucket" => "medium" } }
    }
    if [captcha][solve_time] > 90 {
      mutate { update => { "solve_time_bucket" => "slow" } }
    }
  }

  # Extract date
  date {
    match => ["[captcha][timestamp]", "ISO8601"]
    target => "@timestamp"
  }
}

output {
  elasticsearch {
    hosts => ["elasticsearch:9200"]
    index => "captcha-logs-%{+YYYY.MM.dd}"
  }
}

Field solve_time_bucket dari mutate di atas langsung dipakai di panel Kibana untuk memisahkan solve cepat, sedang, dan lambat.

Index Template Elasticsearch untuk Log CAPTCHA

Tanpa index template, Elasticsearch menebak tipe field secara otomatis — dan sering salah tebak. solve_time bisa saja dikenali sebagai text, bukan float, sehingga agregasi dan sorting jadi tidak bisa dipakai sama sekali. Definisikan tipe field secara eksplisit sejak index pertama dibuat:

{
  "index_patterns": ["captcha-logs-*"],
  "template": {
    "settings": {
      "number_of_shards": 1,
      "number_of_replicas": 0
    },
    "mappings": {
      "properties": {
        "captcha_type": { "type": "keyword" },
        "captcha_id": { "type": "keyword" },
        "error_code": { "type": "keyword" },
        "solve_time": { "type": "float" },
        "poll_count": { "type": "integer" },
        "target_url": { "type": "keyword" },
        "level": { "type": "keyword" },
        "message": { "type": "text" }
      }
    }
  }
}

Terapkan template ini sebelum index harian pertama dibuat, supaya otomatis berlaku untuk setiap index baru berpola captcha-logs-*.

Dashboard Kibana: Panel yang Wajib Ada

Enam panel ini sudah cukup untuk memantau kesehatan pipeline CAPTCHA sehari-hari, dari tingkat keberhasilan sampai solve yang paling lambat:

Sebelum menganggap dashboard ini siap dipakai tim on-call, cek dulu:

• Keenam panel di atas menampilkan data, bukan "No results found"
• Field solve_time dan poll_count bertipe numeric di index pattern Kibana, bukan text
• Time filter default mengarah ke @timestamp hasil parsing Logstash, bukan waktu ingest

Query Kibana yang Sering Dipakai

Simpan query ini sebagai saved search di Kibana supaya tim on-call tidak perlu mengetik ulang setiap kali ada insiden:

# Semua error dalam 1 jam terakhir
level:error AND @timestamp:[now-1h TO now]

# Error timeout untuk reCAPTCHA
error_code:TIMEOUT AND captcha_type:recaptcha_v2

# Solve lambat (> 60 detik)
solve_time:>60

# Error untuk target URL tertentu
level:error AND target_url:"example.com"

# Investigasi CAPTCHA ID tertentu
captcha_id:"73519847"

Troubleshooting Setup ELK untuk Log CAPTCHA

Catatan kepatuhan: hindari mencatat data pribadi pengguna (misalnya URL yang membawa parameter identitas) di log CAPTCHA — sejalan dengan prinsip minimalisasi data pada UU Pelindungan Data Pribadi (UU 27/2022).

Pertanyaan Umum

Berapa banyak worker sebelum ELK Stack sepadan dipakai?

Tidak ada angka pasti, tapi begitu Anda menjalankan puluhan thread paralel — misalnya paket STANDARD ($30/bulan, 15 thread) ke atas — pencarian log manual dengan grep mulai terasa lambat. Di bawah volume itu, grep atau journalctl biasa masih cukup.

Bisakah ELK Stack jalan di VPS kecil, atau wajib pakai cluster besar?

Bisa. Untuk volume sampai beberapa juta log per bulan, satu server dengan 4–8 GB RAM cukup untuk menjalankan Elasticsearch single-node, Logstash, dan Kibana sekaligus. Cluster multi-node baru perlu kalau volumenya jauh lebih besar dari itu.

Berapa lama idealnya log CAPTCHA disimpan?

30 hari cukup untuk kebutuhan operasional sehari-hari; perpanjang sampai 90 hari kalau Anda perlu menganalisis tren bulanan. Atur Elasticsearch ILM supaya index lama terhapus otomatis.

Bisa pakai OpenSearch sebagai pengganti Elasticsearch?

Bisa. OpenSearch kompatibel API dengan Elasticsearch, jadi plugin output Logstash, Filebeat, dan OpenSearch Dashboards (versi open-source dari Kibana) bekerja dengan cara yang sama tanpa perlu mengubah pipeline yang sudah Anda bangun.

Bagaimana cara mendeteksi lonjakan error sebelum berdampak ke proses produksi?

Pasang alert Kibana pada panel "Error breakdown" — misalnya saat rasio level:error melewati ambang tertentu dalam interval 5 menit. Ini jauh lebih cepat daripada menunggu laporan manual dari tim.

Mulai Analisis Log CAPTCHA dengan ELK

Cari dan analisis log CAPTCHA Anda sendiri — dapatkan API key CaptchaAI Anda lalu sambungkan worker ke setup ELK di atas.

Panduan terkait:

• Panduan structured logging untuk operasi CAPTCHA
• Monitoring CAPTCHA dengan Datadog
• Tracing pipeline CAPTCHA dengan OpenTelemetry