Análisis de logs de resolución de CAPTCHA con ELK Stack

A escala, resolver CAPTCHA se reduce a tres preguntas que debes responder en segundos: ¿cuál es mi tasa de éxito ahora mismo?, ¿qué error se dispara? y ¿qué tan lenta va la cola? Con unos pocos workers, un grep sobre el log alcanza; con miles de resoluciones al día, no. ELK Stack (Elasticsearch, Logstash y Kibana) recoge esos logs, los indexa y te deja buscarlos, agregarlos y graficarlos para aislar una incidencia antes de que un cliente la reporte.

Qué te muestran tus logs de CAPTCHA

• Piensa en una agencia en Madrid o Ciudad de México que resuelve CAPTCHA para QA de sus flujos de checkout y para monitoreo de precios en marketplaces regionales: el volumen es irregular y el coste mensual en USD es predecible, pero la tasa de resolución no lo es.
• Cuando esa tasa cae, el equipo necesita saber en minutos si el culpable es un tipo de CAPTCHA concreto, una URL objetivo o toda la cola.
• Un panel de ELK responde eso sin abrir un servidor por SSH, siempre que hayas registrado los datos correctos desde el primer día.

Qué campos registrar en cada resolución

• captcha_id — para rastrear un caso puntual de principio a fin.
• captcha_type — para segmentar reCAPTCHA v2 frente a Turnstile.
• solve_time — la latencia real medida en tu lado.
• error_code — el motivo exacto del fallo.
• target_url — para detectar los sitios que fallan más de la cuenta.
• poll_count — cuántas veces sondeaste el resultado antes de resolver.
• Nunca guardes el token de la solución: no aporta valor diagnóstico y sí añade riesgo.

Arquitectura del pipeline de logs

El recorrido es lineal: los workers escriben JSON, Filebeat lo recoge, Logstash lo enriquece y Elasticsearch lo indexa; Kibana consulta encima.

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

Logging estructurado en JSON

El primer eslabón es el worker: en vez de texto libre, emite cada evento como una línea JSON para que Filebeat y Logstash lo parseen sin expresiones regulares frágiles.

Python: emitir cada evento como 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"}

JavaScript (Node.js): el mismo patrón

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

Del log al índice: el pipeline de ingesta

Con los eventos ya en JSON, tres piezas los llevan hasta Elasticsearch. Filebeat vigila el directorio del worker y reenvía cada línea; Logstash parsea el JSON, clasifica la latencia en tramos (fast, medium, slow) y normaliza la marca de tiempo; y la plantilla de índice declara como keyword los campos por los que filtras (tipo, código de error, URL), sin lo cual las consultas se arrastran.

Filebeat: recoger y enviar los logs

# 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: parsear y clasificar

# 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: plantilla de índice

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

Problemas frecuentes y su solución

Paneles de Kibana que valen la pena

• El panel de tasa de éxito es el primero que abres cuando algo se siente raro.
• El desglose de errores y la latencia en el tiempo te dicen por qué.
• Los seis paneles de la tabla siguiente se arman con los campos que ya registras.

Consultas que resuelven incidencias reales

Estas búsquedas en Kibana Query Language son las que usas cuando cae la tasa de resolución: aíslan por tiempo, tipo, latencia o caso concreto.

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

Preguntas frecuentes

¿Cómo detecto en Kibana que la tasa de resolución empezó a caer?

Compara el panel de tasa de éxito en una ventana móvil (la última hora frente a las 24 previas) y cruza el desglose de errores. Si un error_code sube justo cuando baja la tasa, ahí empiezas; si suben todos a la vez, mira primero la latencia y la cola.

¿Vale la pena montar ELK si ya tengo el panel de CaptchaAI?

Depende del volumen. El panel de CaptchaAI muestra el estado de tu cuenta; ELK correlaciona tus resoluciones con tu infraestructura (latencia, URLs, reintentos) y guarda el histórico. A partir de unos miles de resoluciones diarias, esa correlación propia compensa el montaje.

¿Puedo usar OpenSearch en lugar de Elasticsearch?

Sí. OpenSearch es compatible a nivel de API, así que la salida de Logstash, Filebeat y OpenSearch Dashboards funcionan igual. Los ejemplos se aplican sin cambios.

¿Cuánto tiempo conviene conservar los logs y cómo automatizarlo?

30 días para operación diaria y hasta 90 para análisis de tendencias. No borres a mano: configura Index Lifecycle Management (ILM) para eliminar los índices antiguos y controlar el coste de almacenamiento.

Siguiente paso

Centraliza y analiza tus logs de resolución de CAPTCHA de punta a punta: obtén tu clave API de CaptchaAI y levanta tu stack ELK.

Guías relacionadas:

• Logging estructurado para operaciones de CAPTCHA
• Monitoreo con Datadog
• Trazas con OpenTelemetry