ELK Stack ile CAPTCHA çözüm loglarını analiz etme

CAPTCHA işlem hattınızın nerede yavaşladığını görmek istiyorsanız doğru araç grep değil, çözüm loglarını ELK Stack'e (Elasticsearch, Logstash, Kibana) akıtmaktır. Binlerce çözüm işleyen bir e-ticaret QA ekibinde TIMEOUT dalgasını düz metin loglarda aramak dakikalar alır — Kibana'da aynı sorgu saniyeler sürer.

Fark, ölçekte iyice belirginleşir. Onlarca thread eşzamanlı çalışırken her worker'ın ayrı bir dosyaya yazdığı düz metin, hangi sitekey'in yavaşladığını ya da hangi hata kodunun arttığını söylemez. Çözüm loglarını tek bir yerde toplayıp yapılandırdığınızda ise "son bir saatte reCAPTCHA v2 çözüm süresi neden yükseldi?" sorusu tek bir Kibana sorgusuna dönüşür. Aşağıda, worker loglarını uçtan uca ELK'ye taşıyan üretime hazır bir kurulum adım adım anlatılıyor.

Veri akışı: worker'dan Kibana'ya

Loglar tek yönde akar ve her aşama tek bir işi yapar. Her worker satır başına JSON kaydı üretir; Filebeat dosyaları izleyip iletir, Logstash alanları ayrıştırıp zenginleştirir, Elasticsearch indeksler ve saklar, Kibana sorgular ve görselleştirir. Bu ayrım, sonradan bir bileşeni (örneğin Logstash yerine doğrudan Elasticsearch'e yazma) değiştirmenizi kolaylaştırır:

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

Yapılandırılmış loglama (structured logging)

İlk kural: satır başına tek bir JSON nesnesi yazın ki Kibana alanları filtreleyebilsin. Her log satırında en az şu alanlar bulunmalı:

• captcha_id — bir görevi gönderimden çözüme izler; en kritik alandır
• captcha_type — türe göre kırılım (reCAPTCHA v2, Turnstile, GeeTest v3…)
• solve_time ve poll_count — gecikme ve sorgulama maliyeti
• error_code — başarısızlıkları gruplamak için

Çözüm metnini ise asla loglamayın (nedeni için SSS'ye bakın).

Python: JSON log çıktısı

Formatter, logging kaydına eklediğiniz alanları otomatik JSON'a çevirir:

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"}

JavaScript: yapılandırılmış log çıktısı

Node.js worker'larında aynı deseni axios ile kurun; her olay tek satırlık JSON olur:

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 };

Filebeat yapılandırması

Filebeat, worker loglarını izleyip Logstash'e iletir; keys_under_root ile JSON alanları kök seviyeye çıkar:

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

Logstash pipeline'ı

Logstash her satırı ayrıştırır ve türetilmiş alanlar ekler. Aşağıdaki filtre solve_time'a göre bir solve_time_bucket (fast/medium/slow) hesaplar; böylece gecikme dağılımını tek bakışta görürsünüz:

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

Elasticsearch index şablonu

Filtreleyeceğiniz alanları keyword, sayısal alanları (solve_time, poll_count) ise uygun sayısal türle eşleyin — yanlış eşlemeyi sonradan düzeltmek pahalıdır:

{
  "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" }
      }
    }
  }
}

Kibana panelleri

Altı panel, çözüm operasyonunuzun tamamını kapsar:

İpucu: panel adlarını sorgularınızla tutarlı tutun; altı ay sonra hangi panelin neyi ölçtüğünü hatırlamak çok daha kolay olur.

İşe yarayan Kibana sorguları

# All errors in the last hour
level:error AND @timestamp:[now-1h TO now]

# Timeout errors for reCAPTCHA
error_code:TIMEOUT AND captcha_type:recaptcha_v2

# Slow solves (> 60 seconds)
solve_time:>60

# Errors for a specific target URL
level:error AND target_url:"example.com"

# Specific CAPTCHA ID investigation
captcha_id:"73519847"

Uyarı kurma: hata oranı yükseldiğinde haber alın

Panolar canlıyken kimse ekrana bakmıyorsa değer üretmez; asıl kazanç, bir eşik aşıldığında otomatik uyarı almaktır. Kibana'nın uyarı özelliği (ya da ElastAlert gibi bir alternatif) sabit aralıklarla bir sorgu çalıştırır ve sonuç eşiği aştığında Slack, e-posta veya bir webhook tetikler.

Pratikte iki uyarı çoğu ekip için yeterlidir:

• Hata oranı sıçraması: son 5 dakikada level:error sayısı normal taban çizginizin belirgin biçimde üzerine çıktığında. Ani ERROR_WRONG_GOOGLEKEY ya da TIMEOUT dalgaları genelde sitekey değişikliğini veya hedef sayfadaki bir bozulmayı işaret eder.
• Gecikme kayması: ortalama solve_time seçtiğiniz eşiği (örneğin 30 saniye) sürekli aştığında. Bu, worker havuzunuzun darboğaza girdiğinin erken sinyalidir.

Uyarı eşiklerini gözlemlenen taban çizginize göre belirleyin; sabit bir "doğru" değer yoktur ve CAPTCHA türüne göre değişir.

Ölçek ve maliyet: log hacmini thread sayısıyla ilişkilendirin

Log hacminiz eşzamanlı thread sayınıza bağlıdır. CaptchaAI thread bazlı faturalandırır ve her planda çözüm sınırsızdır; bu da TL oynaklığına karşı öngörülebilir bir aylık USD maliyeti sağlar. BASIC ($15/ay, 5 thread) günde birkaç bin satır üretirken, ADVANCE ($90/ay, 50 thread) seviyesinde index'iniz çok daha hızlı büyür — gün başına index desenini ve ILM saklama süresini buna göre planlayın. KVKK açısından target_url gibi alanlarda kişisel veri sızdırmayın; yalnızca yol (path) yazın.

Sorun giderme

Sık sorulan sorular

CAPTCHA çözüm token'ını log'a yazmalı mıyım?

Hayır. Çözümler tek kullanımlık token'lardır; tanısal değerleri yoktur, depolama maliyetini artırır ve KVKK açısından risk yaratır. Yalnızca meta veriyi loglayın: captcha_id, captcha_type, solve_time, error_code.

solve_time alanını neden keyword değil sayısal tür olarak eşlemeliyim?

float olarak eşlerseniz Kibana'da ortalama, yüzdelik dilim ve solve_time:>60 gibi aralık sorguları çalışır; keyword olarak eşlenmiş bir sayıda bu toplamalar imkânsızdır.

Log hacmi büyüyünce hangi CaptchaAI planına yükselmeliyim?

Karar log satırına değil, ihtiyaç duyduğunuz eşzamanlı thread sayısına bağlıdır. Faturalandırma thread bazlıdır ve her planda çözüm sınırsızdır; darboğazda BASIC'ten ($15/ay, 5 thread) STANDARD'a ($30/ay, 15 thread) ya da ADVANCE'e ($90/ay, 50 thread) çıkın. Shard ve ILM ayarlarını da buna göre güncelleyin.

Kibana'da "çözüm başarı oranı" panelini nasıl kurarım?

Bir Metric görselleştirmesi oluşturun, level:info AND message:"Solve success" sorgusuyla başarılı çözümleri sayın ve toplam görev sayısına bölün. "Task submitted" ve "Solve success" olaylarını ayrı saydığınız için oran, yeni loglar geldikçe anlık güncellenir.

Farklı CAPTCHA türlerinin loglarını aynı index'te tutabilir miyim?

Evet. Her kaydın captcha_type alanı olduğu sürece reCAPTCHA v2/v3, Cloudflare Turnstile, GeeTest v3 ve BLS gibi tüm türleri tek bir captcha-logs-* index'inde toplayabilir, panelleri captcha_type'a göre filtreleyebilirsiniz. Ayrı index gerekmez; türe göre kırılım, tek bir alanla sağlanır.

Sonraki adımlar

Çözüm loglarınızı analiz etmeye başlayın: CaptchaAI API anahtarınızı alın ve ELK'yi bugün ayağa kaldırın.

İlgili rehberler:

• Yapılandırılmış loglama deseni
• Datadog ile metrik ve uyarı takibi
• OpenTelemetry ile pipeline izleme