Стек ELK для анализа журнала решения CAPTCHA

Вечером в проде подскочила доля ошибок по reCAPTCHA v2, а разобраться в причине можно только пролистав вручную гигабайты текстовых логов с десятка воркеров — знакомая ситуация для любой команды, которая гоняет конвейер решения CAPTCHA на тысячах задач в час. Стек ELK (Elasticsearch, Logstash, Kibana) закрывает именно эту проблему: структурированные JSON-логи стекаются в единый индекс, а Kibana строит по ним дашборд с долей успешных решений, разбивкой ошибок по коду и графиком задержки — вместо grep по разрозненным файлам на каждом воркере.

Дальше — рабочая связка: структурированный логгер в воркере, Filebeat как транспорт, Logstash для разбора и обогащения событий, Elasticsearch как хранилище и Kibana для дашбордов. Схема ниже показывает путь данных от задачи CaptchaAI до графика.

Архитектура: путь от воркера до дашборда

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

Каждый воркер пишет логи локально в JSON, Filebeat читает файлы и пересылает события в Logstash, тот парсит JSON, добавляет вычисляемые поля (например, «ведро» скорости решения) и кладёт документ в Elasticsearch. Kibana читает индекс и строит панели поверх него — без правок в самом воркере после первичной настройки.

Структурированное логирование в JSON

Первый шаг — перестать писать логи произвольным текстом. Структурированный JSON-формат даёт Logstash и Elasticsearch готовые поля для фильтрации: captcha_id, captcha_type, solve_time, error_code, poll_count. Ниже — рабочий логгер для Python и Node.js, который выводит именно такой формат.

Python: структурированный логгер

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

Логгер оборачивает стандартный logging форматтером, который сериализует запись в JSON и добавляет только те поля, что реально присутствуют в вызове — событие «задача отправлена» и событие «решение получено» несут разный набор атрибутов, и лишних пустых полей в документе не остаётся.

JavaScript: логирование в JSON

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

Версия на Node.js делает то же самое без внешней библиотеки логирования — оборачивает console.log сериализацией JSON с фиксированным набором служебных полей (timestamp, level, service). Для типового воркера на Puppeteer или Playwright этого достаточно, чтобы Filebeat сразу подхватил вывод.

Что логировать, а что нет

Не логируйте сам решённый токен — это одноразовое значение без диагностической ценности, а его накопление в Elasticsearch создаёт лишнюю поверхность для утечки. Логируйте метаданные: ID задачи, тип CAPTCHA, время решения, код ошибки. То же касается target_url — если в URL встречаются query-параметры с персональными данными пользователя, обрежьте их до домена и пути перед записью в лог. Это упрощает аудит на соответствие 152-ФЗ «О персональных данных» (или GDPR для трансграничного трафика). Правило простое: логируйте то, что нужно для диагностики, и ничего сверх этого.

Filebeat: доставка логов в Logstash

Filebeat читает JSON построчно и пересылает события в Logstash, ничего не буферизуя на диске дольше необходимого.

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

Параметр json.keys_under_root: true разворачивает поля JSON в корень документа вместо того, чтобы прятать их в подобъект json — так Kibana сразу видит captcha_type и solve_time как отдельные поля, готовые для фильтров и агрегаций.

Logstash: разбор и обогащение событий

Logstash парсит сырой JSON, вычисляет производные поля и приводит временную метку к формату, который понимает Elasticsearch.

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

Блок mutate добавляет поле solve_time_bucket («fast» / «medium» / «slow») по порогам 30 и 90 секунд — удобный срез для дашборда, чтобы не считать перцентили на лету в каждом запросе. Фильтр date переносит timestamp из тела события в служебное поле @timestamp, которым пользуется вся временная навигация Kibana.

Elasticsearch: шаблон индекса для логов CAPTCHA

Индекс-шаблон задаёт типы полей заранее. Без него Elasticsearch сам угадает тип по первому документу, и captcha_type рискует стать text вместо keyword — а это ломает точную фильтрацию и агрегации по коду ошибки или типу CAPTCHA.

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

На проде с высоким потоком задач добавьте Index Lifecycle Management: ежедневный индекс captcha-logs-* без ILM за пару месяцев разрастётся до сотен мелких шардов. Рабочая политика — несколько дней в «hot», далее в «warm», удаление после 30 дней для операционных логов (90 дней, если нужен анализ трендов). Если кластер развёрнут в европейском регионе — типичный выбор для команд, обслуживающих русскоязычную аудиторию с трансграничным трафиком, — учитывайте это при оценке задержки записи из воркеров, поднятых в других регионах.

Kibana: какие панели дашборда настроить

Шесть панелей закрывают почти все повседневные вопросы о том, что происходит с решением CAPTCHA: доля успеха, разбивка ошибок, задержка, всплески ошибок во времени, самые медленные решения и активность очереди.

Полезные запросы Kibana

Эти запросы закрывают типовые вопросы дежурного инженера при разборе инцидента:

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

Типичные проблемы и их решения

Ниже — частые причины, по которым связка Filebeat → Logstash → Elasticsearch → Kibana не работает как ожидается, и что с ними делать.

Часто задаваемые вопросы

Сколько хранить логи решения CAPTCHA?

30 дней достаточно для операционной работы — расследования инцидентов, разбора всплесков ошибок. Если нужен анализ трендов по месяцам, продлите до 90 дней и настройте ILM, чтобы старые индексы удалялись автоматически, а не копились вручную.

Можно ли использовать OpenSearch вместо Elasticsearch?

Да. OpenSearch API-совместим: тот же выходной плагин Logstash, тот же Filebeat, а вместо Kibana — OpenSearch Dashboards с почти идентичным интерфейсом. Миграция сводится к смене хоста в конфигурации вывода.

Как отделить логи по типу CAPTCHA, если воркер решает несколько типов?

Полагайтесь на поле captcha_type — оно проставляется в момент отправки задачи и однозначно определяет фильтр в Kibana (captcha_type:recaptcha_v2, captcha_type:turnstile и так далее). Заводить отдельный индекс на каждый тип не нужно — это только усложнит запросы, которые сравнивают типы между собой.

Что делать, если Elasticsearch не успевает индексировать поток логов?

Сначала проверьте RPS воркеров и распределение solve_time_bucket — если логов действительно много, увеличьте number_of_shards и добавьте узел данных. Второй частый случай — Logstash без буфера: включите persistent queue, чтобы всплеск нагрузки не терялся при недоступности Elasticsearch.

Как настроить алерт при росте доли ошибок решения CAPTCHA?

Заведите в Kibana Alerting правило на индекс captcha-logs-*: если доля документов level:error за скользящее окно 15 минут превышает выбранный порог (например, 10%), отправляйте уведомление в Slack или на email дежурного. Для более гибких сценариев подойдёт ElastAlert поверх того же индекса.

Что дальше

Разверните эту связку на своих воркерах, начиная со структурированного логгера, и переходите к смежным темам:

• CaptchaAI Quickstart: ваше первое решение CAPTCHA за 5 минут
• Как решить reCAPTCHA v2 через API: пошаговое руководство
• Как решить Cloudflare Turnstile через API
• Как решить GeeTest v3 с помощью API