Уявіть: у четвер увечері на /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-атака |
| Структурна | Співвідношення методів змінюється, endpoint'и | Спам-боти, сканування вразливостей |
| Якісна | 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%. Зв'яжіться з нами для консультації та отримайте безкоштовний аудит вашої системи моніторингу.