تحليل سجلّات حل CAPTCHA باستخدام ELK Stack

حين يتراجع معدل نجاح حل CAPTCHA أو يقفز زمن الاستجابة فجأة، يكون السؤال الأول دائماً: أين المشكلة بالضبط؟ يمنحك ELK Stack — أي Elasticsearch وLogstash وKibana — الإجابة في ثوانٍ، إذ يجمع سجلات كل عملية حل في مكان واحد قابل للبحث والتصفية والتصوّر البياني. بدلاً من تمشيط ملفات نصية متفرقة عبر عشرات العمّال، تحصل على لوحة واحدة تكشف أنماط الأخطاء واتجاهات الكمون ونقاط الاختناق قبل أن تتحوّل إلى انقطاع في الخدمة. يبني هذا الدليل خط مراقبة كاملاً لعمليات CaptchaAI: من تسجيل منظّم بصيغة JSON داخل العامل، مروراً بـ Filebeat وLogstash، وصولاً إلى فهارس Elasticsearch ولوحات Kibana.

بنية خط المراقبة من العامل إلى Kibana

قبل كتابة أي سطر، من المفيد تصوّر مسار البيانات كاملاً. تكتب عمّال CaptchaAI سطور سجل بصيغة JSON، يلتقطها Filebeat ويشحنها إلى Logstash الذي يحلّلها ويثريها، ثم يخزّنها في Elasticsearch حيث تعرضها Kibana في لوحات حيّة:

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

كل مكوّن مسؤول عن مهمة واحدة: العامل ينتج السجل، وFilebeat ينقله، وLogstash يهيكله ويثريه، وElasticsearch يفهرسه، وKibana يحوّله إلى رؤية بصرية. هذا الفصل بين المسؤوليات يجعل استبدال أي طبقة لاحقاً — أو تشغيلها على خادم منفصل عند التوسّع — أمراً بسيطاً.

التسجيل المنظّم بصيغة JSON

المفتاح الحقيقي لأي تحليل لاحق هو أن يكتب العامل سجلات مهيكلة لا نصاً حرّاً. عندما يكون كل حقل — معرّف المهمة، ونوع الكابتشا، وزمن الحل، ورمز الخطأ، وعدد مرات الاستطلاع — حقلاً مستقلاً في JSON، يصبح الاستعلام عنه في Kibana مباشراً بلا تعبيرات نمطية هشّة.

Python: إخراج السجلات بصيغة JSON

يعرّف المثال التالي مُنسّقاً (JSONFormatter) يحوّل كل سجل إلى كائن JSON، ثم يرفق حقولاً إضافية مثل captcha_id وsolve_time عند توفّرها. لاحظ أن الدالة تسجّل البيانات الوصفية فقط ولا تسجّل نص الحل نفسه، لأنه رمز أحادي الاستخدام لا قيمة تشخيصية له:

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: دالة log بسيطة تطبع كائن JSON إلى المخرج القياسي، وتضيف حقل service لتمييز مصدر السجل عند تجميع عدة خدمات في الفهرس نفسه:

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

يقرأ Filebeat ملفات السجل من قرص العامل ويشحنها إلى Logstash. يفعّل الإعداد التالي تحليل JSON عند المصدر عبر keys_under_root حتى تصل الحقول إلى 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"]

معالجة السجلات في Logstash

يستقبل Logstash الأحداث من Filebeat على المنفذ 5044، ثم يمرّرها عبر مرشّحات تحلّل JSON وتضيف حقولاً محسوبة. في المثال التالي نصنّف كل عملية حسب زمنها إلى ثلاث فئات — سريعة ومتوسطة وبطيئة — وهي تصنيفات تسهّل لاحقاً بناء لوحات توزيع الكمون:

# 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

لكي تعمل التصفية والتجميع بكفاءة، يجب أن يعرف Elasticsearch نوع كل حقل مسبقاً. يعيّن القالب التالي الحقول القابلة للتصفية — مثل captcha_type وerror_code — كنوع keyword، ويحجز solve_time كعدد عشري وpoll_count كعدد صحيح. استخدام 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

بعد وصول البيانات، تبني في Kibana مجموعة لوحات تغطّي المؤشرات التي تهمّ فريق التشغيل. اللوحات التالية نقطة انطلاق عملية:

استعلامات جاهزة للتشخيص

عند وقوع حادثة، تختصر بعض الاستعلامات الجاهزة وقت التشخيص إلى النصف. احفظها في 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"

سيناريو عملي: تشخيص ارتفاع مفاجئ في الأخطاء

تخيّل فريق مراقبة أسعار في متجر إلكتروني بمنطقة الخليج يشغّل مساراً لجمع بيانات المنافسين خلال موسم الجمعة البيضاء. مع تضاعف حجم الطلبات، يلاحظ الفريق عبر لوحة «نسبة نجاح الحل» في Kibana هبوطاً حادّاً خلال ساعة واحدة. وبفتح لوحة «توزيع الأخطاء»، يتبيّن أن غالبية الحالات رمزها TIMEOUT على reCAPTCHA v2 ولنطاق واحد بعينه. يؤكّد استعلام سريع مثل error_code:TIMEOUT AND captcha_type:recaptcha_v2 أن المشكلة محصورة في ذلك النطاق لا في الخدمة كلها.

من هنا يصبح القرار مبنيّاً على بيانات: إمّا أن الموقع المستهدف رفع مستوى حمايته، أو أن عدد الـ Threads المتاح لا يكفي لذروة الطلب. ولأن CaptchaAI يعتمد تسعيراً قائماً على عدد الـ Threads المتزامنة مع حلول غير محدودة لكل Thread، فإن رفع التزامن مؤقتاً يكفي لاستيعاب الذروة دون تكلفة إضافية لكل عملية حل. الأهم أن ELK حوّل سؤالاً غامضاً — «لماذا تباطأ المسار؟» — إلى إجابة دقيقة خلال دقائق.

معالجة المشكلات الشائعة

أغلب مشكلات التشغيل تعود إلى عدد محدود من الأسباب. يربط الجدول التالي كل عرَض بسببه الأرجح والإجراء المناسب:

الأسئلة الشائعة

ما الحقول التي يجب أن يسجّلها العامل لكل عملية حل؟

سجّل البيانات الوصفية فقط: معرّف المهمة، ونوع الكابتشا، وزمن الحل، وعدد مرات الاستطلاع، ورمز الخطأ. لا تسجّل نص الحل ذاته؛ فهو رمز أحادي الاستخدام لا يحمل قيمة تشخيصية، وتخزينه يرفع الكلفة ويضيف مخاطر أمنية دون فائدة.

كيف أراقب معدل نجاح الحل لحظياً في Kibana؟

أنشئ لوحة من نوع «مقياس» تحسب نسبة أحداث level:info AND message:"Solve success" إلى إجمالي المحاولات ضمن نافذة زمنية متحرّكة، واضبط تحديثها التلقائي كل بضع دقائق لتحصل على مؤشر حيّ لصحة المسار قبل أن يشتكي المستخدمون.

هل يؤثّر التسجيل المكثّف على أداء مسار الحل؟

التسجيل المنظّم بصيغة JSON خفيف الأثر ما دام يكتب إلى المخرج القياسي أو ملف محلي ويُشحن بشكل غير متزامن عبر Filebeat. تجنّب فقط الكتابة المتزامنة إلى الشبكة داخل مسار الحل نفسه حتى لا تضيف زمناً إلى كل طلب.

كيف أميّز بين مشكلة في الخدمة ومشكلة في الموقع المستهدف؟

قارن رمز الخطأ ونطاق target_url. إذا تركّزت الأخطاء في نطاق واحد بعينه فالغالب أن الموقع رفع حمايته؛ أمّا إذا امتدّت عبر جميع النطاقات فالمشكلة أقرب إلى بيئة التشغيل أو بيانات الاعتماد أو حصّة الـ Threads.

الخطوات التالية

• ابدأ سريعاً مع CaptchaAI: حلّ أول كابتشا في خمس دقائق
• حلّ reCAPTCHA v2 عبر الـ API خطوة بخطوة
• حلّ Cloudflare Turnstile باستخدام الـ API
• حلّ GeeTest v3 باستخدام الـ API

أدلة ذات صلة

• دليل البدء السريع مع CaptchaAI
• مراقبة أسعار التجارة الإلكترونية عند ظهور اختبارات CAPTCHA