Создание пользовательских оповещений CaptchaAI с помощью PagerDuty

Голый мониторинг без маршрутизации бесполезен: если алерт по балансу CaptchaAI улетает в общий Slack-канал, который никто не читает по ночам, о проблеме первыми узнают клиенты, а не дежурный. PagerDuty решает именно это — превращает событие мониторинга в инцидент с приоритетом, эскалацией и контекстом для инженера, а не просто ещё одну строку в ленте.

Стратегия оповещений: что считать критичным инцидентом

Не каждое отклонение заслуживает того, чтобы будить дежурного среди ночи. Разделите сигналы по критичности заранее — иначе PagerDuty быстро превратится в источник alert fatigue, а не защиты от него.

Логика простая: всё, что напрямую останавливает поток решений (баланс кончился, воркеры легли), заслуживает немедленной страницы дежурного. Всё, что деградирует постепенно (растёт очередь, растёт задержка), можно завести как инцидент низкого приоритета и разобрать в рабочее время. Такое разделение — не формальность, а единственный способ сохранить доверие к системе оповещений: если каждый алерт равнозначен, дежурный быстро научится их игнорировать.

Python: интеграция с PagerDuty Events API v2

Ниже — рабочая реализация поверх PagerDuty Events API v2: обёртка над trigger/resolve/acknowledge и монитор, который считает долю ошибок в скользящем окне на 5 минут и проверяет баланс через res.php.

import os
import time
import hashlib
import requests
from datetime import datetime

API_KEY = os.environ["CAPTCHAAI_API_KEY"]
PAGERDUTY_ROUTING_KEY = os.environ["PAGERDUTY_ROUTING_KEY"]

session = requests.Session()


class CaptchaPagerDuty:
    EVENTS_URL = "https://events.pagerduty.com/v2/enqueue"

    def __init__(self, routing_key):
        self.routing_key = routing_key

    def trigger(self, summary, severity="error", source="captcha-pipeline",
                details=None, dedup_key=None):
        """Trigger a new PagerDuty incident."""
        payload = {
            "routing_key": self.routing_key,
            "event_action": "trigger",
            "payload": {
                "summary": summary,
                "severity": severity,  # critical, error, warning, info
                "source": source,
                "timestamp": datetime.utcnow().isoformat() + "Z",
                "custom_details": details or {}
            }
        }

        if dedup_key:
            payload["dedup_key"] = dedup_key

        resp = requests.post(self.EVENTS_URL, json=payload, timeout=10)
        resp.raise_for_status()
        return resp.json()

    def resolve(self, dedup_key):
        """Resolve an existing incident."""
        payload = {
            "routing_key": self.routing_key,
            "event_action": "resolve",
            "dedup_key": dedup_key
        }
        resp = requests.post(self.EVENTS_URL, json=payload, timeout=10)
        resp.raise_for_status()
        return resp.json()

    def acknowledge(self, dedup_key):
        """Acknowledge an existing incident."""
        payload = {
            "routing_key": self.routing_key,
            "event_action": "acknowledge",
            "dedup_key": dedup_key
        }
        resp = requests.post(self.EVENTS_URL, json=payload, timeout=10)
        resp.raise_for_status()
        return resp.json()


pagerduty = CaptchaPagerDuty(PAGERDUTY_ROUTING_KEY)


class CaptchaMonitor:
    def __init__(self):
        self.error_window = []  # (timestamp, is_error)
        self.window_size = 300  # 5 minutes in seconds

    def record_solve(self, success):
        now = time.time()
        self.error_window.append((now, not success))
        # Prune old entries
        self.error_window = [
            (t, e) for t, e in self.error_window
            if now - t < self.window_size
        ]

    @property
    def error_rate(self):
        if not self.error_window:
            return 0.0
        errors = sum(1 for _, e in self.error_window if e)
        return errors / len(self.error_window)

    def check_balance(self):
        resp = session.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY, "action": "getbalance", "json": 1
        })
        data = resp.json()
        if data.get("status") != 1:
            return None
        return float(data["request"])

    def run_checks(self):
        """Run all monitoring checks and trigger alerts."""
        # Check balance
        balance = self.check_balance()
        if balance is not None:
            if balance < 2:
                pagerduty.trigger(
                    summary=f"CaptchaAI balance critically low: ${balance:.2f}",
                    severity="critical",
                    dedup_key="captcha-balance-critical",
                    details={"balance": balance, "threshold": 2}
                )
            elif balance < 10:
                pagerduty.trigger(
                    summary=f"CaptchaAI balance low: ${balance:.2f}",
                    severity="warning",
                    dedup_key="captcha-balance-warning",
                    details={"balance": balance, "threshold": 10}
                )
            else:
                # Resolve if balance recovered
                try:
                    pagerduty.resolve("captcha-balance-critical")
                    pagerduty.resolve("captcha-balance-warning")
                except Exception:
                    pass  # No incident to resolve

        # Check error rate
        rate = self.error_rate
        if rate > 0.20:
            total = len(self.error_window)
            errors = sum(1 for _, e in self.error_window if e)
            pagerduty.trigger(
                summary=f"CaptchaAI error rate {rate:.0%} "
                        f"({errors}/{total} in 5 min)",
                severity="error",
                dedup_key="captcha-error-rate-high",
                details={
                    "error_rate": round(rate, 3),
                    "total_tasks": total,
                    "failed_tasks": errors,
                    "window_seconds": self.window_size
                }
            )
        elif rate < 0.05 and len(self.error_window) > 10:
            try:
                pagerduty.resolve("captcha-error-rate-high")
            except Exception:
                pass


monitor = CaptchaMonitor()

# After each solve:
# monitor.record_solve(success=True)

# Run checks every 60 seconds:
# while True:
#     monitor.run_checks()
#     time.sleep(60)

JavaScript: тот же паттерн для Node.js

Та же логика на Node.js — для команд, у которых пайплайн решения CAPTCHA уже живёт в JavaScript-стеке.

const axios = require("axios");

const API_KEY = process.env.CAPTCHAAI_API_KEY;
const PD_ROUTING_KEY = process.env.PAGERDUTY_ROUTING_KEY;
const PD_EVENTS_URL = "https://events.pagerduty.com/v2/enqueue";

class PagerDutyAlerter {
  constructor(routingKey) {
    this.routingKey = routingKey;
  }

  async trigger(summary, severity = "error", details = {}, dedupKey = null) {
    const payload = {
      routing_key: this.routingKey,
      event_action: "trigger",
      payload: {
        summary,
        severity,
        source: "captcha-pipeline",
        timestamp: new Date().toISOString(),
        custom_details: details,
      },
    };
    if (dedupKey) payload.dedup_key = dedupKey;

    const resp = await axios.post(PD_EVENTS_URL, payload, { timeout: 10000 });
    return resp.data;
  }

  async resolve(dedupKey) {
    await axios.post(PD_EVENTS_URL, {
      routing_key: this.routingKey,
      event_action: "resolve",
      dedup_key: dedupKey,
    }, { timeout: 10000 });
  }
}

const alerter = new PagerDutyAlerter(PD_ROUTING_KEY);

class CaptchaHealthMonitor {
  constructor(windowMs = 300000) {
    this.results = [];
    this.windowMs = windowMs;
  }

  record(success) {
    this.results.push({ time: Date.now(), success });
    const cutoff = Date.now() - this.windowMs;
    this.results = this.results.filter((r) => r.time > cutoff);
  }

  get errorRate() {
    if (this.results.length === 0) return 0;
    const errors = this.results.filter((r) => !r.success).length;
    return errors / this.results.length;
  }

  async checkAndAlert() {
    // Balance check
    try {
      const resp = await axios.get("https://ocr.captchaai.com/res.php", {
        params: { key: API_KEY, action: "getbalance", json: 1 },
      });
      if (resp.data.status === 1) {
        const balance = parseFloat(resp.data.request);
        if (balance < 2) {
          await alerter.trigger(
            `CaptchaAI balance critically low: $${balance.toFixed(2)}`,
            "critical",
            { balance },
            "captcha-balance-critical"
          );
        } else if (balance < 10) {
          await alerter.trigger(
            `CaptchaAI balance low: $${balance.toFixed(2)}`,
            "warning",
            { balance },
            "captcha-balance-warning"
          );
        } else {
          await alerter.resolve("captcha-balance-critical").catch(() => {});
          await alerter.resolve("captcha-balance-warning").catch(() => {});
        }
      }
    } catch (err) {
      console.error("Balance check failed:", err.message);
    }

    // Error rate check
    const rate = this.errorRate;
    if (rate > 0.2 && this.results.length > 10) {
      await alerter.trigger(
        `CaptchaAI error rate: ${(rate * 100).toFixed(1)}%`,
        "error",
        { errorRate: rate, totalTasks: this.results.length },
        "captcha-error-rate"
      );
    } else if (rate < 0.05 && this.results.length > 10) {
      await alerter.resolve("captcha-error-rate").catch(() => {});
    }
  }
}

const monitor = new CaptchaHealthMonitor();

// Run checks every 60 seconds
setInterval(() => monitor.checkAndAlert(), 60000);

module.exports = { monitor, alerter };

Чек-лист настройки PagerDuty

Прежде чем подключать код, настройте сам сервис в PagerDuty:

1. Создайте в PagerDuty отдельный сервис для «CaptchaAI Pipeline».
2. Подключите к сервису интеграцию Events API v2.
3. Скопируйте routing key в переменную окружения PAGERDUTY_ROUTING_KEY.
4. Настройте политику эскалации: дежурный → тимлид → руководитель.
5. Настройте каналы уведомлений: push, SMS, звонок.
6. Добавьте окна обслуживания на плановые простои.

Для команд, где дежурные разнесены по часовым поясам между Москвой, Алматы и Киевом, шаг 4 особенно важен: без чёткой многоуровневой эскалации ночная смена в одном городе рискует ждать реакции от инженера, который в это время спит в другом.

И не кладите в custom_details персональные данные пользователей — только технические метрики (баланс, частота ошибок, глубина очереди). Это особенно актуально, если логи инцидентов подпадают под 152-ФЗ «О персональных данных» или под GDPR для части аудитории. Такой подход заодно упрощает ретроспективу инцидентов: в карточке остаётся только то, что нужно для диагностики, а не персональные данные, которые потом придётся объяснять службе безопасности.

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

Частые вопросы

Как проверить интеграцию, не дожидаясь реального сбоя баланса?

Вызовите pagerduty.trigger() вручную с тестовым summary и отдельным dedup_key вроде captcha-test-alert, убедитесь, что карточка дошла до нужного канала уведомлений, затем вызовите resolve() с тем же ключом и проверьте, что инцидент закрылся. Прогоните такой тест перед тем, как полагаться на алерты в бою, и повторяйте его после любых правок политики эскалации.

Как выстроить эскалацию, если дежурные раскиданы по часовым поясам?

Задайте многоуровневую политику эскалации в PagerDuty: если дежурный не подтверждает инцидент за заданное время, вызов автоматически уходит следующему по цепочке — в том числе инженеру в другом часовом поясе. Это стандартная практика для распределённых CIS-команд, где инженеры физически сидят в Москве, Алматы и Киеве одновременно.

Чем отличаются trigger, acknowledge и resolve?

trigger создаёт новый инцидент. acknowledge останавливает уведомления, но оставляет инцидент открытым — кто-то уже занимается проблемой. resolve полностью закрывает инцидент. Все три действия используют один и тот же dedup_key, поэтому именно он связывает вызовы между собой в единую цепочку.

Можно ли подключить PagerDuty к Datadog или Zabbix вместо прямой интеграции?

Да. У Datadog есть готовая интеграция с PagerDuty из коробки; для Zabbix обычно нужен отдельный webhook-модуль или скрипт-посредник. Прямая интеграция через Events API, как в этом руководстве, даёт больше контроля над форматом summary и custom_details, но требует немного больше кода на вашей стороне.

Что дальше

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