Automated Metric Alerts with AI: как узнать о проблеме раньше пользователей

Большинство сбоев в production обнаруживают пользователи, а не инженерная команда. Время между поломкой и первым тикетом в поддержку — десятки минут. За это время продукт теряет конверсию, пользователи уходят, NPS падает. Автоматические алерты сокращают окно обнаружения до 2–5 минут.

Статья описывает полный стек: от anomaly detection до конкретных промптов, которые генерируют правила алертинга. Три инструмента (Grafana, PostHog, custom-решения), готовые конфигурации, типичные ошибки.

Почему статические пороги не работают для продуктовых метрик

Классический подход: “если error rate > 5%, отправь алерт”. Работает для инфраструктурных метрик (CPU, память, диск). Ломается на продуктовых метриках по трём причинам.

Сезонность. Конверсия в покупку в e-commerce в понедельник утром и в пятницу вечером отличается в разы. Статический порог будет либо пропускать реальные проблемы, либо генерировать ложные срабатывания каждый день.

Тренды. Продукт растёт. Абсолютное число ошибок растёт вместе с трафиком. Порог, установленный месяц назад, перестаёт быть релевантным.

Распределение. Среднее время отклика API = 200ms. Но p99 = 4 секунды. Статический порог на среднее пропустит деградацию, которая затрагивает 1% самых активных пользователей.

Anomaly detection решает все три проблемы. Вместо фиксированного числа система строит динамический baseline и реагирует на отклонения от него.

Anomaly detection: три подхода

Statistical (Z-score, IQR)

Самый простой. Вычисляем среднее и стандартное отклонение за скользящее окно. Если текущее значение отклоняется больше чем на N стандартных отклонений, срабатывает алерт.

import numpy as np

def detect_anomaly_zscore(values: list[float], threshold: float = 3.0) -> bool:
    """Z-score anomaly detection на скользящем окне."""
    if len(values) < 30:
        return False

    mean = np.mean(values[:-1])
    std = np.std(values[:-1])

    if std == 0:
        return False

    z_score = abs(values[-1] - mean) / std
    return z_score > threshold

Плюсы: прост в реализации, минимум ресурсов. Минусы: не учитывает сезонность. Подходит для метрик с относительно стабильным поведением (error rate, latency p99).

Seasonal decomposition (STL, Prophet)

Разбивает временной ряд на три компонента: тренд, сезонность, остаток. Алерт срабатывает на аномалии в остатке.

from statsmodels.tsa.seasonal import STL

def detect_anomaly_stl(
    series,  # pd.Series с DatetimeIndex
    period: int = 24,  # часовая сезонность
    threshold: float = 3.0
) -> bool:
    """STL decomposition + Z-score на остатке."""
    stl = STL(series, period=period, robust=True)
    result = stl.fit()

    residual = result.resid
    mean_r = residual.mean()
    std_r = residual.std()

    if std_r == 0:
        return False

    latest_residual = residual.iloc[-1]
    z_score = abs(latest_residual - mean_r) / std_r
    return z_score > threshold

Подходит для метрик с выраженной периодичностью: DAU, конверсия, revenue. Требует минимум 2-3 полных цикла данных для обучения.

ML-based (Isolation Forest, Autoencoders)

Для многомерных аномалий, когда отдельные метрики в норме, но их комбинация указывает на проблему.

from sklearn.ensemble import IsolationForest

def build_anomaly_detector(features_matrix):
    """Isolation Forest для многомерного anomaly detection."""
    model = IsolationForest(
        n_estimators=100,
        contamination=0.01,  # ожидаем 1% аномалий
        random_state=42
    )
    model.fit(features_matrix)
    return model

# features: [error_rate, latency_p99, conversion_rate, active_users]
# Каждая метрика в норме по отдельности,
# но комбинация "latency растёт + conversion падает" = проблема

Используйте ML-based подход, когда простые методы дают слишком много ложных срабатываний или пропускают сложные паттерны деградации.

Grafana: алерты с anomaly detection

Grafana Alerting (начиная с v9) поддерживает multi-dimensional alerts, contact points и notification policies. Базовая настройка занимает 15 минут.

Настройка alert rule для error rate

# grafana-alert-rule.yaml
apiVersion: 1
groups:
  - orgId: 1
    name: product-health
    folder: Alerts
    interval: 1m
    rules:
      - uid: error-rate-anomaly
        title: "Error Rate Anomaly"
        condition: C
        data:
          - refId: A
            relativeTimeRange:
              from: 3600  # последний час
              to: 0
            datasourceUid: prometheus
            model:
              expr: |
                sum(rate(http_requests_total{status=~"5.."}[5m]))
                /
                sum(rate(http_requests_total[5m]))
              intervalMs: 60000
          - refId: B
            relativeTimeRange:
              from: 604800  # последняя неделя для baseline
              to: 0
            datasourceUid: prometheus
            model:
              expr: |
                avg_over_time(
                  (sum(rate(http_requests_total{status=~"5.."}[5m]))
                   /
                   sum(rate(http_requests_total[5m]))
                  )[7d:1h]
                )
          - refId: C
            datasourceUid: "__expr__"
            model:
              type: math
              expression: "$A > ($B * 3)"  # 3x от baseline
        for: 5m
        labels:
          severity: critical
          team: backend
        annotations:
          summary: "Error rate {{ $values.A }} превышает baseline {{ $values.B }} в 3+ раза"

Notification policy с эскалацией

# grafana-notification-policy.yaml
apiVersion: 1
policies:
  - orgId: 1
    receiver: default-slack
    group_by: ['alertname', 'team']
    group_wait: 30s
    group_interval: 5m
    repeat_interval: 4h
    routes:
      - receiver: pagerduty-critical
        matchers:
          - severity = critical
        continue: true
        group_wait: 0s
      - receiver: slack-warnings
        matchers:
          - severity = warning
        group_wait: 1m

Grafana позволяет строить алерты на основе PromQL-запросов с математическими выражениями. Можно сравнивать текущее значение с baseline за прошлую неделю, с тем же временем вчера или с percentile за месяц.

PostHog: алерты на продуктовые события

PostHog подходит для product-level метрик: конверсия по воронкам, retention, feature adoption. PostHog поддерживает встроенные алерты на Insights.

Настройка алерта на воронку

1. Создайте Insight типа Funnel: signup -> onboarding_complete -> first_action
2. В настройках Insight выберите Alert threshold
3. Установите условие: conversion rate < baseline - 2 стандартных отклонения

PostHog вычисляет baseline автоматически на основе данных за последние 30 дней.

Алерт через API

import httpx

POSTHOG_HOST = "https://app.posthog.com"
POSTHOG_API_KEY = "phx_..."
PROJECT_ID = "12345"

async def create_posthog_alert(
    insight_id: int,
    threshold_type: str,  # "absolute" | "relative"
    threshold_value: float,
    notification_targets: list[dict]
):
    """Создание алерта на PostHog Insight через API."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{POSTHOG_HOST}/api/projects/{PROJECT_ID}/alerts/",
            headers={"Authorization": f"Bearer {POSTHOG_API_KEY}"},
            json={
                "insight": insight_id,
                "name": f"Alert for insight {insight_id}",
                "threshold": {
                    "type": threshold_type,
                    "value": threshold_value,
                    "configuration": {
                        "compare_to": "previous_period",
                        "direction": "decrease"
                    }
                },
                "notification_targets": notification_targets,
                "enabled": True
            }
        )
        return response.json()

# Пример: алерт если конверсия упала на 20% относительно прошлого периода
await create_posthog_alert(
    insight_id=42,
    threshold_type="relative",
    threshold_value=-0.20,
    notification_targets=[
        {"type": "slack", "channel": "#product-alerts"},
        {"type": "email", "address": "[email protected]"}
    ]
)

Custom-решение: Python + Prometheus + Slack

Когда Grafana или PostHog не покрывают специфический сценарий, custom-сервис на 200 строк решает задачу. Типичный кейс: алерт на комбинацию метрик из разных источников.

Архитектура

┌─────────────┐     ┌─────────────────┐     ┌──────────┐
│ Prometheus   │────▶│  Alert Service   │────▶│  Slack   │
│ PostHog API  │────▶│  (Python)        │────▶│  PagerDuty│
│ Custom DB    │────▶│                  │────▶│  Telegram │
└─────────────┘     └─────────────────┘     └──────────┘
                           │
                    ┌──────┴──────┐
                    │  Rules DB   │
                    │  (YAML/DB)  │
                    └─────────────┘

Сервис алертинга

import asyncio
from dataclasses import dataclass
from enum import Enum
import httpx
import yaml

class Severity(Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"

@dataclass
class AlertRule:
    name: str
    metric_query: str
    source: str  # "prometheus" | "posthog" | "custom"
    condition: str  # "above" | "below" | "anomaly"
    threshold: float | None
    window_minutes: int
    severity: Severity
    channels: list[str]
    cooldown_minutes: int = 30

class MetricAlertService:
    def __init__(self, config_path: str):
        self.rules = self._load_rules(config_path)
        self.last_fired: dict[str, float] = {}
        self.prometheus_url = "http://prometheus:9090"
        self.slack_webhook = "https://hooks.slack.com/..."

    def _load_rules(self, path: str) -> list[AlertRule]:
        with open(path) as f:
            config = yaml.safe_load(f)
        return [AlertRule(**rule) for rule in config["rules"]]

    async def check_rule(self, rule: AlertRule) -> bool:
        """Проверка одного правила."""
        if rule.source == "prometheus":
            value = await self._query_prometheus(rule.metric_query)
        elif rule.source == "posthog":
            value = await self._query_posthog(rule.metric_query)
        else:
            value = await self._query_custom(rule.metric_query)

        if value is None:
            return False

        if rule.condition == "above":
            return value > rule.threshold
        elif rule.condition == "below":
            return value < rule.threshold
        elif rule.condition == "anomaly":
            history = await self._get_history(
                rule.source, rule.metric_query, rule.window_minutes
            )
            return detect_anomaly_zscore(history + [value])
        return False

    async def _query_prometheus(self, query: str) -> float | None:
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{self.prometheus_url}/api/v1/query",
                params={"query": query}
            )
            data = resp.json()
            results = data.get("data", {}).get("result", [])
            if not results:
                return None
            return float(results[0]["value"][1])

    async def run_check_cycle(self):
        """Один цикл проверки всех правил."""
        for rule in self.rules:
            triggered = await self.check_rule(rule)
            if triggered and self._cooldown_passed(rule.name, rule.cooldown_minutes):
                await self._send_alert(rule)
                self.last_fired[rule.name] = asyncio.get_event_loop().time()

    async def _send_alert(self, rule: AlertRule):
        message = f"[{rule.severity.value.upper()}] {rule.name}"
        for channel in rule.channels:
            if channel == "slack":
                await self._send_slack(message)
            elif channel == "pagerduty":
                await self._send_pagerduty(rule)

    def _cooldown_passed(self, rule_name: str, cooldown_min: int) -> bool:
        last = self.last_fired.get(rule_name, 0)
        now = asyncio.get_event_loop().time()
        return (now - last) > cooldown_min * 60

Конфигурация правил

# alert-rules.yaml
rules:
  - name: "API Error Rate Spike"
    metric_query: 'sum(rate(http_requests_total{status=~"5.."}[5m])) / sum(rate(http_requests_total[5m]))'
    source: prometheus
    condition: anomaly
    threshold: null
    window_minutes: 60
    severity: critical
    channels: ["slack", "pagerduty"]
    cooldown_minutes: 15

  - name: "Signup Conversion Drop"
    metric_query: "funnel/signup-to-activation"
    source: posthog
    condition: below
    threshold: 0.15  # конверсия ниже 15%
    window_minutes: 1440  # за сутки
    severity: warning
    channels: ["slack"]
    cooldown_minutes: 360

  - name: "Payment Success Rate"
    metric_query: 'sum(rate(payments_total{status="success"}[10m])) / sum(rate(payments_total[10m]))'
    source: prometheus
    condition: below
    threshold: 0.95  # ниже 95% успешных платежей
    window_minutes: 30
    severity: critical
    channels: ["slack", "pagerduty"]
    cooldown_minutes: 10

  - name: "LLM Response Quality"
    metric_query: 'avg(llm_response_score{model="gpt-5.4"}[15m])'
    source: prometheus
    condition: below
    threshold: 0.7
    window_minutes: 60
    severity: warning
    channels: ["slack"]
    cooldown_minutes: 60

AI-промпты для генерации правил алертинга

Вместо ручного написания каждого правила LLM генерирует конфигурации на основе описания продукта. Три промпта покрывают 90% сценариев.

Промпт 1: генерация правил из описания продукта

Ты — SRE-инженер. На основе описания продукта сгенерируй YAML-конфигурацию
для системы алертинга.

Продукт: {product_description}
Доступные метрики: {metrics_list}
Источники данных: {datasources}

Для каждого правила определи:
- name: понятное название
- metric_query: точный запрос к источнику данных
- condition: above/below/anomaly
- threshold: числовое значение (null для anomaly)
- severity: info/warning/critical
- window_minutes: размер окна для оценки
- cooldown_minutes: минимальный интервал между повторными алертами

Правила:
1. Critical — только для метрик, напрямую влияющих на revenue или доступность
2. Warning — деградация, которая может стать critical без вмешательства
3. Для каждой critical метрики добавь warning на более мягком пороге
4. Anomaly detection — для метрик с выраженной сезонностью

Формат: YAML, совместимый с alert-rules.yaml

Промпт 2: анализ существующих алертов и устранение пробелов

Проанализируй текущую конфигурацию алертов и найди пробелы.

Текущие правила:
{current_rules_yaml}

Архитектура системы:
{system_architecture}

Инциденты за последние 3 месяца:
{incident_history}

Определи:
1. Какие типы инцидентов не были бы обнаружены текущими правилами?
2. Какие правила генерируют ложные срабатывания (на основе паттернов инцидентов)?
3. Какие правила нужно добавить?
4. Какие пороги нужно скорректировать?

Для каждого найденного пробела предложи конкретное правило в формате YAML.

Промпт 3: автоматическая калибровка порогов

На основе исторических данных метрики рассчитай оптимальный порог для алерта.

Метрика: {metric_name}
Данные за 30 дней (JSON): {metric_data_json}
Известные инциденты в этом периоде: {incidents}
Текущий порог: {current_threshold}
Текущее количество ложных срабатываний в неделю: {false_positives_per_week}

Требования:
- Максимум 2 ложных срабатывания в неделю
- Все известные инциденты должны быть обнаружены
- Учти дневную и недельную сезонность

Верни:
1. Рекомендуемый порог
2. Рекомендуемый window_minutes
3. Ожидаемое количество ложных срабатываний
4. Какие из известных инцидентов будут пропущены (если есть)

Эти промпты работают с GPT-5.4, Claude, Gemini. LLM-генерация правил сокращает время настройки алертинга с нескольких дней до нескольких часов. Каждое сгенерированное правило нужно валидировать на исторических данных перед деплоем.

12 правил для production-алертинга

Правила, проверенные на практике. Каждое предотвращает конкретную проблему.

1. Cooldown обязателен. Без cooldown один инцидент генерирует десятки алертов. Минимум: 10 минут для critical, 30 минут для warning.

2. Алертируйте на симптомы, не на причины. “Конверсия упала на 30%” лучше, чем “CPU > 80%”. Пользователю безразличен CPU.

3. Каждый алерт требует action. Если на алерт нечего делать, удалите его. Информационные алерты отправляйте в отдельный канал с пометкой INFO.

4. Группируйте по сервису. Каскадный сбой генерирует алерты от 10 сервисов. Группировка по service в notification policy предотвращает информационный шторм.

5. Разделяйте каналы по severity. Critical в PagerDuty (будит ночью). Warning в Slack (разбирают утром). Info в дашборд (смотрят на планёрке).

6. Используйте for (pending period). Алерт срабатывает только если условие выполняется N минут подряд. Исключает одиночные спайки. Рекомендация: 3-5 минут для critical, 10-15 для warning.

7. Один алерт — один owner. Если непонятно, кто реагирует, алерт бесполезен. Label team в каждом правиле + routing в notification policy.

8. Пересматривайте пороги ежемесячно. Продукт меняется. Метрики дрейфуют. Алерт, настроенный три месяца назад, может стать нерелевантным.

9. Считайте alert fatigue. Если команда получает больше 5 алертов в день, она начинает их игнорировать. Мониторьте количество срабатываний и false positive rate.

10. Тестируйте алерты на исторических данных. Перед деплоем нового правила прогоните его на данных за последний месяц. Все ли известные инциденты обнаружены? Сколько ложных срабатываний?

11. Документируйте runbook. Каждый алерт содержит ссылку на runbook: что проверить, как починить, кого эскалировать. В аннотацию Grafana-алерта добавляйте поле runbook_url.

12. Мониторьте сам алертинг. Dead man’s switch: сервис периодически отправляет “I’m alive” алерт. Если он не пришёл, система алертинга сломана.

# Dead man's switch в Grafana
- uid: alerting-health-check
  title: "Alerting System Health"
  condition: A
  data:
    - refId: A
      model:
        expr: 'up{job="alert-service"}'
  for: 5m
  labels:
    severity: critical
  annotations:
    summary: "Alert service is down — alerts are not being evaluated"
    runbook_url: "https://wiki.internal/runbooks/alerting-down"

Интеграция с LLM Observability

Если продукт использует LLM, стандартных метрик недостаточно. HTTP 200 не означает, что модель вернула полезный ответ. Мониторинг качества LLM-ответов требует отдельного уровня алертинга.

Langfuse (подробнее в руководстве по LLM Observability) предоставляет метрики, которые можно экспортировать в Prometheus:

# Экспорт Langfuse scores в Prometheus
from prometheus_client import Gauge
from langfuse import Langfuse

langfuse = Langfuse()
llm_quality = Gauge("llm_response_quality", "LLM response quality score", ["model", "prompt_name"])
llm_cost = Gauge("llm_cost_per_request", "LLM cost per request in USD", ["model"])
llm_latency = Gauge("llm_latency_seconds", "LLM response latency", ["model"])

async def export_langfuse_metrics():
    """Периодический экспорт метрик из Langfuse в Prometheus."""
    traces = langfuse.fetch_traces(limit=100, order_by="timestamp.desc")

    for trace in traces.data:
        if trace.scores:
            for score in trace.scores:
                llm_quality.labels(
                    model=trace.metadata.get("model", "unknown"),
                    prompt_name=trace.metadata.get("prompt_name", "unknown")
                ).set(score.value)

        if trace.total_cost:
            llm_cost.labels(
                model=trace.metadata.get("model", "unknown")
            ).set(trace.total_cost)

Алерты на LLM-метрики:

rules:
  - name: "LLM Quality Degradation"
    metric_query: 'avg(llm_response_quality{prompt_name="main-assistant"}[30m])'
    source: prometheus
    condition: below
    threshold: 0.6
    window_minutes: 30
    severity: warning
    channels: ["slack"]

  - name: "LLM Cost Spike"
    metric_query: 'sum(rate(llm_cost_per_request[1h])) * 3600'
    source: prometheus
    condition: above
    threshold: 50  # $50/час
    window_minutes: 60
    severity: critical
    channels: ["slack", "pagerduty"]

Отказоустойчивость алертинга и circuit breaker

Сервис алертинга сам может упасть. Два паттерна решают проблему.

Dead man’s switch (описан выше): внешний сервис проверяет, что система алертинга жива.

Circuit breaker на уровне интеграций: если Slack API не отвечает, алерт отправляется в fallback-канал (email, Telegram). Подробнее о паттерне circuit breaker в статье про Deno Edge Functions.

from dataclasses import dataclass, field
from time import time

@dataclass
class CircuitBreaker:
    failure_threshold: int = 3
    reset_timeout_sec: int = 60
    failures: int = 0
    last_failure: float = 0
    state: str = "closed"  # closed | open | half-open

    def record_failure(self):
        self.failures += 1
        self.last_failure = time()
        if self.failures >= self.failure_threshold:
            self.state = "open"

    def record_success(self):
        self.failures = 0
        self.state = "closed"

    def can_execute(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if time() - self.last_failure > self.reset_timeout_sec:
                self.state = "half-open"
                return True
            return False
        return True  # half-open: пробуем один запрос

# Использование
slack_circuit = CircuitBreaker(failure_threshold=3, reset_timeout_sec=120)

async def send_alert_with_fallback(message: str):
    if slack_circuit.can_execute():
        try:
            await send_slack(message)
            slack_circuit.record_success()
        except Exception:
            slack_circuit.record_failure()
            await send_telegram(message)  # fallback
    else:
        await send_telegram(message)  # circuit open, используем fallback

Чеклист внедрения

Минимальный набор алертов для запуска:

Порядок внедрения: сначала error rate и latency (Grafana + Prometheus, один час), затем payment/revenue метрики, затем продуктовые метрики через PostHog, затем LLM-метрики при необходимости. Дополнительные правила — через AI-промпты выше. Ежемесячный ревью: удалять алерты без полезных срабатываний, рекалибровать остальные.

Разница между “пользователи жалуются” и “мы уже чиним” определяет, вернутся ли эти пользователи завтра.

Нужна помощь с настройкой мониторинга? Я помогаю стартапам внедрять AI-решения и строить продукты — belov.works.