ELK-Stack für die CAPTCHA-Lösungsprotokollanalyse

„Warum steigt die Fehlerquote seit heute Nacht, und welche target_url ist betroffen?" – auf diese Frage gibt Ihnen grep über verteilte Worker-Logs keine brauchbare Antwort. Der ELK-Stack aus Elasticsearch, Logstash und Kibana schon: Er verwandelt die JSON-Logs Ihrer CAPTCHA-Worker in durchsuchbare Kennzahlen, aus denen Sie Fehlermuster, Latenztrends und Timeout-Häufungen in Sekunden ablesen. Dieser Leitfaden zeigt die vollständige Pipeline – vom strukturierten Log-Eintrag bis zum Kibana-Dashboard.

Wann sich der ELK-Stack lohnt

Solange Sie ein paar Hundert CAPTCHA-Lösungen pro Tag abwickeln, reichen Konsolen-Logs und ein Terminal. Ab dem Punkt, an dem mehrere Worker parallel laufen, sich die Threads Ihres CaptchaAI-Plans gleichzeitig füllen und ein einzelner Ausfall in der Log-Flut untergeht, brauchen Sie Aggregation statt Textsuche. Typische Auslöser aus der Praxis:

• Eine bestimmte target_url läuft plötzlich gehäuft in Timeouts.
• Ein error_code taucht nur zu Stoßzeiten auf.
• Die Lösungszeit für reCAPTCHA v2 driftet über Tage langsam nach oben.

Genau diese Fragen beantwortet ein Dashboard, keine Grep-Pipeline.

Die Pipeline im Überblick

Der Datenfluss ist geradlinig: Ihre Worker schreiben strukturierte JSON-Zeilen, Filebeat sammelt sie ein, Logstash parst und reichert sie an, Elasticsearch indiziert sie, und Kibana macht sie sichtbar.

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

Strukturiertes Logging als Fundament

Jede spätere Auswertung ist nur so gut wie das Log-Format an der Quelle. Schreiben Sie deshalb keine freien Textzeilen, sondern ein JSON-Objekt pro Ereignis mit festen Feldern:

• captcha_id – eindeutige Kennung der Aufgabe;
• captcha_type – der Typ, etwa reCAPTCHA v2 oder Cloudflare Turnstile;
• solve_time – gemessene Lösungszeit in Sekunden;
• error_code – Fehlercode bei Abbruch;
• target_url und poll_count – Ziel-URL und Anzahl der Abfrage-Zyklen.

So bleiben die Felder in Elasticsearch filterbar, statt in einem Fließtext zu verschwinden.

Python – JSON-Log-Ausgabe

Der folgende Formatter hängt die Zusatzfelder nur an, wenn sie vorhanden sind, und gibt jede Zeile als kompaktes JSON aus. solve_captcha protokolliert den vollständigen Lebenszyklus einer Aufgabe: Übermittlung, jeden Abfrage-Zyklus (Polling) und das Endergebnis mit gemessener Lösungszeit.

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 – dasselbe Log-Schema für Node.js-Worker

Wer die Lösung aus einem Node.js-Worker heraus aufruft, sollte identische Feldnamen verwenden, damit beide Sprachwelten im selben Index landen und dieselben Dashboards bedienen.

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

Logs mit Filebeat einsammeln

Filebeat liest die JSON-Zeilen direkt von der Platte und übergibt sie an Logstash. Mit keys_under_root: true landen Ihre Felder auf oberster Ebene, statt in einem verschachtelten json-Objekt zu stecken – das erspart später eine Parsing-Stufe.

# 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: parsen und anreichern

Logstash ist die Stelle, an der aus rohen Feldern nützliche Dimensionen werden. Hier wird die solve_time in Buckets (fast, medium, slow) einsortiert und der Zeitstempel des Workers auf @timestamp gemappt, damit Kibana korrekt nach Ereigniszeit sortiert.

# 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-Template

Ohne explizites Mapping rät Elasticsearch die Feldtypen – und macht aus filterbaren Werten oft ein text-Feld, auf dem sich schlecht aggregieren lässt. Legen Sie deshalb ein Template an, das captcha_type, error_code und target_url als keyword und solve_time als float festschreibt.

{
  "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-Dashboards aufbauen

Mit sauberen Feldern lassen sich die entscheidenden Panels in Minuten zusammenklicken. Diese sechs decken den Alltag beim Betrieb einer CAPTCHA-Pipeline ab:

Nützliche Kibana-Abfragen

Für die schnelle Analyse im Discover-Tab sparen ein paar gespeicherte Abfragen viel Zeit. Setzen Sie bei der letzten Abfrage Ihre eigene Beispiel-Domain ein (etwa staging.example-app.test).

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

DSGVO: Was in CAPTCHA-Logs gehört – und was nicht

Für Leser im DACH-Raum ist die Log-Frage auch eine Datenschutzfrage. Felder wie target_url und – je nach Deployment – die Client-IP-Adresse Ihrer Worker können nach DSGVO als personenbezogene Daten gelten. Zwei praktische Konsequenzen:

• Prüfen Sie Rechtsgrundlage und Aufbewahrungsdauer Ihrer Log-Daten.
• Protokollieren Sie niemals den gelösten Token selbst.

Das gelöste CAPTCHA ist ein Einmal-Token ohne diagnostischen Wert, das ohnehin nach kurzer Zeit abläuft; es zu speichern erhöht nur Speicherkosten und Angriffsfläche. Für die Fehlersuche genügen Metadaten: captcha_id, captcha_type, solve_time, poll_count und error_code. Wer sensible Ziel-URLs verarbeitet, kann diese vor der Indizierung in Logstash pseudonymisieren.

Fehlerbehebung

Häufige Fragen

Welche Felder gehören in ein strukturiertes CAPTCHA-Log?

captcha_id, captcha_type, solve_time, poll_count, error_code und die Ziel-URL. Damit lassen sich Erfolgsquote, Latenz und Fehlerursachen auswerten. Den gelösten Token selbst nie mitloggen.

Brauche ich Logstash, oder reicht Filebeat direkt an Elasticsearch?

Für reines Weiterleiten genügt Filebeat direkt. Logstash lohnt sich, sobald Sie berechnete Felder wie die solve_time-Buckets, das Datums-Mapping oder eine Pseudonymisierung brauchen – also für die Anreicherung.

Wie lange sollte ich die Logs aufbewahren?

30 Tage decken den operativen Betrieb ab, 90 Tage reichen für Trendanalysen. Richten Sie ILM ein, damit alte Indizes automatisch gelöscht werden – das begrenzt Speicherkosten und Datenschutzrisiko zugleich.

Funktioniert der Stack auch mit OpenSearch statt Elasticsearch?

Ja. OpenSearch ist weitgehend API-kompatibel; Filebeat, das Logstash-Output-Plugin und die OpenSearch Dashboards ersetzen Kibana ohne Änderung am Log-Schema. Ein Umstieg berührt Ihre Worker nicht.

Verwandte Leitfäden

• Strukturiertes Logging für CAPTCHA-Vorgänge
• Grafana-Dashboard-Vorlagen für CaptchaAI
• CAPTCHA-Lösungsraten mit Prometheus und Grafana überwachen