ELK Stack cho phân tích nhật ký giải CAPTCHA

CaptchaAI worker của bạn chạy hàng trăm request mỗi giờ, một số bắt đầu trả về TIMEOUT lúc nửa đêm, và tất cả những gì bạn có là vài file log rải rác trên từng máy? Đó chính xác là lúc grep hết tác dụng. ELK Stack — Elasticsearch, Logstash, Kibana — biến log giải CAPTCHA thành dữ liệu tìm kiếm được: lọc theo captcha_type, error_code hay target_url trong vài giây thay vì đọc log tay từng máy. Nhiều đội automation/QA tại các công ty outsourcing ở TP.HCM và Hà Nội, vận hành hàng chục worker song song, dùng đúng mô hình dưới đây để theo dõi pipeline giải CAPTCHA trong môi trường production.

Kiến trúc: log chảy từ worker đến Kibana thế nào

Bốn khối chính tạo nên pipeline này:

• Worker ghi log JSON có cấu trúc ra stdout hoặc file.
• Filebeat theo dõi file log và forward sang Logstash.
• Logstash parse JSON, tính thêm trường như solve_time_bucket rồi đẩy vào Elasticsearch.
• Kibana là nơi bạn truy vấn, dựng dashboard và đặt cảnh báo.

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

Ghi log JSON có cấu trúc ngay trong worker

Log dạng văn bản tự do không lọc được. Chỉ khi log theo trường cố định — captcha_id, captcha_type, solve_time, error_code, poll_count — Elasticsearch mới đánh index và Kibana mới tổng hợp được chính xác. Hai ví dụ dưới đây ghi log ở bước gửi task, poll kết quả và báo lỗi timeout.

Python - Đầu ra nhật ký JSON

JSONFormatter dưới đây tự gắn captcha_id, captcha_type, solve_time, error_code, poll_count vào mỗi dòng log khi bạn truyền chúng qua tham số extra:

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 – Ghi nhật ký có cấu trúc

Cùng logic đó viết bằng Node.js, dùng console.log với JSON thay vì một logger riêng:

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

Đưa log vào Elasticsearch: Filebeat, Logstash và index template

Ba mảnh còn lại của pipeline chạy ở tầng hạ tầng, không đụng tới code worker: Filebeat thu thập, Logstash xử lý, index template quyết định Elasticsearch lưu dữ liệu theo kiểu gì.

Filebeat: theo dõi và forward log

Filebeat là agent nhẹ chạy cạnh worker, theo dõi thư mục log và forward từng dòng JSON sang Logstash qua cổng 5044. Cấu hình tối thiểu chỉ cần khai báo đường dẫn file và output:

# 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: parse JSON và gắn nhãn tốc độ giải

Logstash nhận dữ liệu từ Filebeat, parse trường message thành JSON có cấu trúc, rồi tự tính solve_time_bucket (fast/medium/slow) để bạn lọc nhanh các lần giải chậm mà không phải viết truy vấn range mỗi lần. Cuối pipeline, dữ liệu được đẩy vào Elasticsearch theo index đặt tên theo ngày.

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

Index template cho log CAPTCHA

Khai báo template trước khi log đầu tiên chảy vào, để Elasticsearch ánh xạ đúng kiểu dữ liệu — keyword cho các trường cần lọc, float cho solve_time — thay vì tự đoán và biến mọi thứ thành text, nguyên nhân phổ biến nhất khiến truy vấn chậm về sau.

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

Dashboard Kibana nên theo dõi những gì

Sáu panel dưới đây đủ để một đội vận hành nắm được sức khỏe pipeline giải CAPTCHA mà không cần mở terminal:

Các câu truy vấn Lucene dùng hằng ngày

Lưu sẵn các query sau trong Kibana Discover hoặc Saved Search — chúng trả lời gần hết câu hỏi debug thường gặp khi pipeline gặp sự cố:

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

Xử lý sự cố ELK thường gặp

Bốn lỗi dưới đây chiếm phần lớn ticket khi mới triển khai ELK cho log CAPTCHA:

• Log không xuất hiện trong Kibana — thường do Filebeat không gửi log; kiểm tra log của Filebeat và xác nhận đường dẫn file khớp với cấu hình.
• Lỗi parse JSON — do có dòng log không phải JSON lẫn trong file; bật json.keys_under_root trong Filebeat và sửa lại định dạng output của logger.
• Quá nhiều index — do tạo index theo ngày nhưng không có ILM; thiết lập Index Lifecycle Management (ILM) và giữ log 30 ngày.
• Truy vấn chậm — do thiếu ánh xạ kiểu keyword; dùng kiểu keyword cho các trường cần lọc, không dùng text.

Câu hỏi thường gặp

Log CAPTCHA nên giữ bao lâu thì đủ?

30 ngày cho log vận hành hằng ngày là đủ để debug sự cố gần đây. Giữ 90 ngày nếu bạn cần phân tích xu hướng theo quý. Đặt ILM tự xóa index cũ để chi phí lưu trữ không tăng vô hạn theo thời gian.

Log chứa target_url có cần lưu ý gì theo Nghị định 13/2023/NĐ-CP không?

Nếu target_url hoặc trường log khác chứa dữ liệu cá nhân (ví dụ URL gắn ID người dùng), hãy áp dụng nguyên tắc tối thiểu hóa dữ liệu và giới hạn quyền truy cập theo Nghị định 13/2023/NĐ-CP về bảo vệ dữ liệu cá nhân. Bài viết này không phải tư vấn pháp lý — chỉ là gợi ý kỹ thuật để bạn thiết kế schema log an toàn hơn.

Team nhỏ nên tự host ELK hay dùng managed service?

Nếu bạn chỉ chạy vài worker, một dịch vụ managed như Elastic Cloud hoặc AWS OpenSearch Service tiết kiệm thời gian vận hành hơn tự host. Tự host ELK chỉ hợp lý khi khối lượng log đủ lớn để chi phí managed service vượt chi phí nhân sự duy trì cluster.

Có nên ghi lại token giải CAPTCHA vào log không?

Không. Token là chuỗi dùng một lần, không có giá trị chẩn đoán, lưu chúng chỉ tốn thêm dung lượng mà không giúp bạn debug thêm được gì — lại tiềm ẩn rủi ro bảo mật. Chỉ log metadata: ID, loại CAPTCHA, thời gian giải, trạng thái.

Một index có chứa log của nhiều loại CAPTCHA cùng lúc được không?

Có. Trường captcha_type (reCAPTCHA v2, Cloudflare Turnstile, GeeTest v3...) đủ để Kibana lọc riêng từng loại ngay trong cùng index captcha-logs-*. Bạn chỉ cần tách index theo loại khi khối lượng log của một loại lớn tới mức ảnh hưởng hiệu năng truy vấn.

Bước tiếp theo

Tìm kiếm và phân tích log giải CAPTCHA của bạn — lấy API key CaptchaAI và dựng pipeline ELK theo các bước trên.

Hướng dẫn liên quan:

• Ghi log có cấu trúc cho pipeline CAPTCHA
• Giám sát CaptchaAI bằng Datadog
• Theo dõi phân tán với OpenTelemetry