Pilha ELK para análise de log de resolução de CAPTCHA

Um pico de erros no CAPTCHA aparece de madrugada — foi timeout do provedor, uma chave sem saldo ou instabilidade de rede? Com grep espalhado em arquivos de log por worker, essa resposta leva minutos, um arquivo de cada vez, e ainda depende de lembrar em qual servidor o worker estava rodando naquela hora.

No ELK Stack (Elasticsearch, Logstash, Kibana), a mesma investigação vira uma consulta de poucos segundos: filtre por error_code, agrupe por captcha_type e leia a tendência de latência direto no painel, sem entrar em nenhum servidor manualmente. Este guia monta o pipeline de ponta a ponta:

• Logs em JSON estruturado nos workers (Python e Node.js).
• Coleta com o Filebeat e enriquecimento no Logstash.
• Indexação no Elasticsearch com mapeamento correto desde o início.
• Painéis e consultas prontas no Kibana para investigar incidentes.

Arquitetura do pipeline de logs de CAPTCHA

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

O worker grava logs em JSON; o Filebeat encaminha, o Logstash processa e enriquece, e o Elasticsearch indexa para consulta no Kibana. Cada etapa faz um trabalho só — isso facilita achar onde um problema começou quando um worker novo entra na frota e os logs dele não aparecem como esperado no painel.

Estruture os logs de CAPTCHA em JSON antes de tudo

Sem formato estruturado, campos como error_code, solve_time e captcha_type simplesmente não existem para o Elasticsearch indexar depois. Os exemplos abaixo emitem exatamente os mesmos campos em Python e em JavaScript — mantendo os nomes idênticos entre os dois workers, o índice único do Elasticsearch consegue misturar os dois idiomas sem duplicar mapeamento.

Python: log estruturado em 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: o mesmo padrão em Node.js

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

Envie os logs com o Filebeat

O Filebeat só lê os arquivos *.log do worker e encaminha cada linha JSON ao Logstash — ele não interpreta nem transforma o conteúdo, então qualquer erro de parsing aparece na etapa seguinte, não aqui. Se o worker roda em vários containers ou instâncias, o mesmo filebeat.yml funciona em todos; basta configurar um processor de metadados de host para diferenciar a origem de cada linha no índice.

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

Processe os logs no pipeline do Logstash

O Logstash recebe o payload do Filebeat, faz o parse do JSON e calcula um campo derivado (solve_time_bucket) para facilitar filtros por faixa de latência antes de gravar no Elasticsearch. É também o lugar certo para adicionar novos campos calculados — por exemplo, marcar is_beta_type: true quando captcha_type for CaptchaFox (beta), Friendly Captcha (beta) ou Lemin (beta), para separar esses volumes nos painéis sem mexer no código do worker.

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

Defina o índice no Elasticsearch

O template abaixo fixa o tipo de cada campo do índice captcha-logs-* — sem isso, o Elasticsearch tenta adivinhar o tipo sozinho e error_code/captcha_type acabam mapeados como texto livre, o que quebra os filtros exatos usados nos painéis a seguir. Se a maior parte do seu tráfego é do Brasil, hospedar o cluster numa região como sa-east-1 da AWS reduz a latência de escrita; para volumes menores, number_of_shards: 1 (já usado abaixo) evita overhead desnecessário — só aumente ao passar de alguns GB por dia no í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" }
      }
    }
  }
}

Monte os painéis de logs de CAPTCHA no Kibana

Com o índice mapeado, estes seis painéis cobrem o que costuma importar no dia a dia: taxa de sucesso, tipo de erro predominante e latência. Comece só com eles — dá para adicionar painéis por captcha_type ou por região do worker depois, conforme a operação crescer.

Consultas prontas para investigar logs de CAPTCHA

As consultas abaixo usam os mesmos campos do índice captcha-logs-* — copie e ajuste o intervalo de tempo:

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

Solução de problemas

Dica: se nada disso resolver, confirme primeiro que o próprio worker está gravando em /var/log/captcha-worker/*.log — boa parte dos "problemas no Kibana" começa numa etapa anterior do pipeline.

Antes de ativar em produção

Confira estes pontos antes de apontar o pipeline inteiro para produção:

• Índice captcha-logs-* criado com o template de mapeamento aplicado.
• ILM configurado com a retenção definida (30 ou 90 dias).
• Os seis painéis do Kibana funcionando com dados reais, não só com os exemplos deste guia.
• Nenhum campo sensível (texto da solução, dados pessoais) chegando ao índice.

Perguntas frequentes

Vale a pena montar um ELK Stack só para logs de CAPTCHA, ou o Datadog já resolve?

Depende do volume. Se já paga por SaaS de observabilidade, agregar os logs lá custa menos esforço. Acima de milhões de eventos por mês, um Elasticsearch próprio sai mais barato.

A partir de qual volume de tarefas compensa migrar do grep para o ELK?

Com dezenas de tarefas por dia, grep ainda resolve bem. A partir de alguns milhares por dia, centralizar no Elasticsearch já compensa.

Por quanto tempo devo guardar os logs de CAPTCHA, considerando a LGPD?

30 dias bastam para casos operacionais; 90 dias se precisar de análise de tendência. Configure o ILM do Elasticsearch para excluir índices antigos. Como os logs não devem conter dados pessoais nem o texto da solução (próxima pergunta), o cuidado de conformidade é limitar o acesso ao índice.

Preciso registrar o texto da solução do CAPTCHA nos logs?

Não. É um token de uso único sem valor de diagnóstico — registrá-lo só aumenta custo e risco. Grave apenas metadados: ID, tipo, latência e status.

Coloque o ELK para rodar

Consulte seus logs de CAPTCHA em segundos — obtenha sua chave de API e configure o pipeline ELK.

Guias relacionados:

• Registro estruturado
• Monitoramento com Datadog
• Rastreamento com OpenTelemetry