Pile ELK pour l'analyse des journaux de résolution CAPTCHA

La stack ELK transforme les logs bruts de votre pipeline CAPTCHA en tableaux de bord interrogeables : un pic d'erreurs TIMEOUT, une dérive de latence ou une URL cible qui échoue en boucle deviennent visibles en quelques secondes, sans jamais rouvrir un fichier .log à la main. Elasticsearch indexe chaque résolution, Logstash normalise le flux entrant et Kibana affiche les tendances — taux de réussite, répartition des erreurs, résolutions les plus lentes. Ce guide branche CaptchaAI sur cette chaîne, du worker qui émet du JSON structuré jusqu'au dashboard Kibana prêt à l'emploi.

Architecture : du worker CAPTCHA à Kibana

Le flux est unidirectionnel. Chaque worker écrit ses logs en JSON sur disque ; Filebeat les collecte et les pousse vers Logstash, qui parse, enrichit et route les événements vers Elasticsearch ; Kibana lit l'index et sert les visualisations. Chaque étage reste remplaçable : vous pouvez héberger le cluster sur OVHcloud, Scaleway ou une région AWS eu-west-3 (Paris) sans toucher au code des workers.

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

Journalisation structurée des résolutions

Tout commence par le format des logs. Une ligne exploitable, c'est un objet JSON par événement, avec des champs typés : captcha_id, captcha_type, solve_time, error_code, poll_count et target_url. Le texte libre est précisément ce qui rend grep inutilisable à l'échelle. Les deux implémentations ci-dessous — Python et Node.js — émettent exactement cette structure sur la sortie standard, que Filebeat récupère ensuite.

Python : émettre chaque résolution en JSON

Le JSONFormatter sérialise chaque enregistrement et n'ajoute un champ que s'il est présent, ce qui garde les lignes compactes. La fonction solve_captcha journalise trois moments clés : la soumission, le succès (avec la latence mesurée) et l'échec ou le timeout.

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

Node.js : journalisation structurée

La version JavaScript suit la même logique avec une fonction log minimaliste. Le champ service permet de distinguer plusieurs workers dans un même index.

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

Configuration de Filebeat

Filebeat surveille le répertoire des logs et transmet chaque ligne à Logstash. Comme les workers écrivent déjà du JSON, activez le décodage natif pour que les champs remontent à la racine du document plutôt que dans un blob message.

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

Pipeline Logstash : parser et enrichir

Logstash parse le JSON, puis enrichit chaque événement. Ici, un champ calculé solve_time_bucket classe les résolutions en fast, medium (> 30 s) ou slow (> 90 s) — pratique pour segmenter les dashboards sans requête complexe. Le filtre date aligne l'horodatage du worker sur @timestamp, ce qui rend les graphes temporels fiables même en cas de retard d'ingestion.

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

Modèle d'index Elasticsearch

Le mapping est décisif pour la performance. Déclarez captcha_type, error_code et target_url en keyword (filtrage exact et agrégations), solve_time en float et message en text pour la recherche plein texte. L'index journalier captcha-logs-AAAA.MM.JJ facilite ensuite la purge par ILM.

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

Tableaux de bord Kibana

Six panneaux couvrent l'essentiel de l'exploitation quotidienne. Construisez-les une fois, puis épinglez-les dans un tableau de bord partagé avec l'équipe.

Requêtes KQL utiles

Gardez ces requêtes sous la main pour les investigations ad hoc dans Discover.

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

Conservation des logs et RGPD

Les logs de résolution peuvent contenir des données à caractère personnel — typiquement dans target_url, quand l'URL embarque un identifiant utilisateur ou un e-mail. Appliquez le principe de minimisation : ne journalisez que les métadonnées nécessaires au diagnostic (identifiant, type, latence, statut), tronquez ou hachez les paramètres sensibles de l'URL, et ne conservez jamais le token de résolution. Côté durée, 30 jours couvrent l'exploitation courante ; au-delà, l'analyse de tendance se satisfait d'agrégats sur 90 jours. Documentez ce choix dans votre registre de traitements et automatisez la suppression avec Elasticsearch ILM plutôt que de purger les index à la main.

Dépannage

FAQ

Faut-il journaliser le contenu du token de résolution ?

Non. Le token est à usage unique et n'a aucune valeur diagnostique une fois consommé. Le stocker gonfle l'index et crée un risque de sécurité inutile : ne conservez que les métadonnées (identifiant, type, latence, statut).

Quelle durée de conservation choisir pour rester conforme au RGPD ?

Trente jours pour les logs opérationnels, quatre-vingt-dix jours si vous faites de l'analyse de tendance. Appliquez la minimisation des données, documentez la durée dans votre registre de traitements et laissez Elasticsearch ILM supprimer automatiquement les index expirés.

Comment configurer une alerte Kibana sur un pic d'erreurs ?

Créez une règle dans Kibana Alerting basée sur une requête level:error, avec un seuil (par exemple plus de 20 erreurs sur 5 minutes) et un connecteur Slack ou e-mail. Le champ error_code permet de router chaque alerte selon le type d'échec.

Peut-on envoyer les logs directement à Elasticsearch sans Logstash ?

Oui, via le module Elasticsearch de Filebeat, ce qui suffit pour un pipeline simple. Mais dès que vous ajoutez des champs calculés comme solve_time_bucket ou que vous devez normaliser plusieurs formats de logs, Logstash reste l'étage d'enrichissement le plus souple à maintenir.

Comment suivre la latence P90 par type de CAPTCHA ?

Ajoutez une agrégation de percentiles sur solve_time dans Kibana, segmentée par captcha_type. Le champ solve_time_bucket calculé dans Logstash permet en plus de filtrer rapidement les résolutions lentes (> 90 s).

Prochaines étapes

Vous avez la chaîne complète : logs JSON côté worker, ingestion Filebeat, enrichissement Logstash, index Elasticsearch et dashboards Kibana. Pour la mettre en place, récupérez votre clé API CaptchaAI et instrumentez votre premier worker.

Guides associés :

• La journalisation structurée des opérations CAPTCHA
• La supervision avec Datadog
• Le traçage distribué avec OpenTelemetry