验证码日志分析：ELK Stack 实战教程

验证码识别任务量一旦到每天几万条，grep 加 tail -f 就查不过来了。ELK Stack（Elasticsearch、Logstash、Kibana）能把日志变成可检索、可聚合、可视化的数据——错误码分布、耗时趋势、异常任务，几秒钟就能定位。

整体架构

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

Worker 只管把 JSON 打到标准输出，采集、解析、入库、展示全交给 ELK。

结构化日志：先定好该记什么字段

核心字段固定为 captcha_id、captcha_type、solve_time、error_code、target_url、poll_count，多语言 worker 要统一命名，避免同时出现 solveTime 和 solve_time。日志只记元数据，不写识别结果（token）原文——详见后面常见问题。

Python 示例：输出结构化 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 示例：结构化日志输出

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 采集日志

Worker 把 JSON 写到日志文件后，用 Filebeat 实时转发给 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"]

keys_under_root: true 让 JSON 字段平铺到顶层，而不是塞进嵌套对象，后面查询更方便。

编写 Logstash 管道

Logstash 再解析一遍 JSON、按耗时分桶打标签，并把时间戳标准化成 @timestamp：

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

solve_time_bucket 这类派生字段很实用，不用每次手写范围查询，按 fast/medium/slow 筛选即可。

Elasticsearch 索引模板

字段按 keyword 精确匹配还是按 text 分词，这一步没配好会直接拖慢后面的查询：

{
  "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 仪表板

常见的排错和容量规划面板：

面板	类型	逻辑
识别成功率	Metric	`level:info AND message:"Solve success"` 占比
错误分布	饼图	按 `error_code` 分组统计
耗时趋势	折线图	`solve_time` 平均值
错误趋势	柱状图	每 5 分钟统计 `level:error`
最慢任务	数据表	按 `solve_time` 降序前 10
队列活跃度	面积图	按 `message` 计数

常用查询示例

排错时最常用的几条查询：

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

常见故障排查

问题	原因	处理方式
Kibana 看不到日志	Filebeat 没发送	查看 Filebeat 日志，确认 `paths` 匹配到文件
JSON 解析报错	混入非 JSON 行	加 `json.keys_under_root`，检查日志输出格式
索引数量暴涨	按天建索引没配 ILM	配置 ILM，30 天自动清理
worker 依赖装不上/很慢	国内访问 PyPI/npm 官方源不稳定	换清华 TUNA 等国内镜像源

常见问题

验证码日志应该保留多久？

日常排错留 30 天，趋势分析或容量规划建议留 90 天，用 Elasticsearch ILM 自动过期删除即可。

worker 用了好几种语言，日志字段名怎么保证一致？

统一约定一份字段清单，各语言照清单实现、命名风格一致，否则 Kibana 聚合会跟着出错。

应该把识别结果（token）也记到日志里吗？

不建议。token 是一次性凭证，记了没有排错价值，只会增加存储成本和安全风险。

Kibana 面板一直显示不出最新数据，是什么原因？

先确认 Filebeat 有没有发送日志、Logstash 有没有解析报错；都正常的话，多半是面板时间范围选窄了。

下一步

相关指南：

用于验证码解决日志分析的 ELK Stack

整体架构

结构化日志：先定好该记什么字段

Python 示例：输出结构化 JSON 日志

JavaScript 示例：结构化日志输出

用 Filebeat 采集日志

编写 Logstash 管道

Elasticsearch 索引模板

搭建 Kibana 仪表板

常用查询示例

常见故障排查

常见问题

验证码日志应该保留多久？

worker 用了好几种语言，日志字段名怎么保证一致？

应该把识别结果（token）也记到日志里吗？

Kibana 面板一直显示不出最新数据，是什么原因？

下一步

AWS Lambda + CaptchaAI：无服务器验证码解决

用于 CaptchaAI Worker 部署的 Ansible Playbook

使用 AWS SNS 和 CaptchaAI 构建事件驱动的验证码解决方案

验证码解决基础设施的蓝绿部署

Azure Functions + CaptchaAI：云集成

使用 CaptchaAI 在 Python 中构建验证码求解队列

整体架构

结构化日志：先定好该记什么字段

Python 示例：输出结构化 JSON 日志

JavaScript 示例：结构化日志输出

用 Filebeat 采集日志

编写 Logstash 管道

Elasticsearch 索引模板

搭建 Kibana 仪表板

常用查询示例

常见故障排查

常见问题

验证码日志应该保留多久？

worker 用了好几种语言，日志字段名怎么保证一致？

应该把识别结果（token）也记到日志里吗？

Kibana 面板一直显示不出最新数据，是什么原因？

下一步

相关文章

AWS Lambda + CaptchaAI：无服务器验证码解决

用于 CaptchaAI Worker 部署的 Ansible Playbook

使用 AWS SNS 和 CaptchaAI 构建事件驱动的验证码解决方案

验证码解决基础设施的蓝绿部署

Azure Functions + CaptchaAI：云集成

使用 CaptchaAI 在 Python 中构建验证码求解队列