זיהוי חריגות תעבורה: EWMA, Prometheus, התראות

זיהוי חריגות תעבורה: EWMA, Prometheus, התראות

תארו לעצמכם: בערב יום חמישי, /api/checkout מקבל לפתע זינוק של 500 שגיאות. הניטור נשאר שקט כי הסף הסטטי לא נחצה — השיא הוא רק 30% מהמקסימום היומי (רק 300 בקשות בשנייה במקום 1000). שלוש שעות לאחר מכן אתם שמים לב לבעיה מתלונות של לקוחות. או תרחיש אחר: סקריפט גירוד מתחיל להכות בקטלוג, ושולח 10,000 בקשות בדקה — מגבלת קצב רגילה לא מופעלת כי התעבורה מפוזרת על פני כתובות IP. אנו פורסים מערכת זיהוי חריגות תעבורה שמאתרת זינוקים כאלה תוך 5 שניות ושולחת התראה ל-Slack. ללא אזעקות שווא בשיאים עונתיים. יש לנו ניסיון של 10+ שנים בבניית מערכות ניטור ויישמנו מעל 50 פרויקטים של זיהוי חריגות. לדוגמה, עם לקוח שיש לו 50,000 מבקרים ייחודיים ביום, צמצמנו את זמן הזיהוי מ-3 שעות ל-5 שניות. החיסכון מהיישום יכול להגיע לעשרות אלפי דולרים בשנה על ידי צמצום זמן השבתה. קבלו ייעוץ מהנדס כדי לכוונן ספים לתעבורה שלכם.

איך פועל זיהוי חריגות תעבורה

הגלאי שלנו משלב שתי שיטות סטטיסטיות: EWMA (ממוצע נע משוקלל מעריכית) ו-z-score על חלון נע. הראשונה מסתגלת למגמות, השנייה תופסת חריגים חדים. לפי EWMA, זוהי שיטת החלקה מעריכית. אנו עוטפים זאת לשירות שאוסף מדדים בזמן אמת דרך Redis, מנתח כל דקה, ושולח התראות עם הפחתה אוטומטית. זמן הזיהוי מופחת ב-99% בהשוואה לניתוח לוגים ידני.

למה מערכות ניטור סטנדרטיות לא מספיקות

Zabbix או Nagios טיפוסיים משתמשים בספים סטטיים. הם נותנים אזעקות שווא בשיאי תעבורה בשעות מכירות ומפספסים דליפות איטיות. המערכת שלנו אדפטיבית: היא מתכווננת לעונתיות ולמגמות. אנו מכווננים רגישות באופן אישי לפרופיל התעבורה שלכם, תוך שימוש בנתונים היסטוריים מ-2–4 שבועות. לדוגמה, במהלך Black Friday, אנו מעלים ספים אוטומטית כדי למנוע התראות שווא. זה אפשר ללקוח אחד לצמצם את זמן התגובה לאירועים משעתיים ל-10 דקות.

איך בנויה מערכת זיהוי החריגות

אנו משלבים EWMA ו-z-score על חלון נע. EWMA מגיב מהר יותר למגמות, z-score תופס טוב יותר חריגים חדים. עבור קווי בסיס ארוכי טווח אנו משתמשים ב-Prometheus עם כללים מבוססי אחוזונים. הגלאי כתוב ב-Python אך ניתן להעבירו ל-Go עבור מערכות בעומס גבוה (10,000+ RPS).

מה נחשב לחריגה?

אנו מזהים שלושה סוגי חריגות:

• נפח: RPS, רוחב פס, מספר כתובות IP ייחודיות עולה בחדות (למשל, מ-100 ל-5000 בדקה).
• מבני: יחס שיטות HTTP משתנה (למשל, עלייה חדה ב-GET כשהנורמה היא 70/30 GET/POST), עלייה בבקשות לנקודות קצה ספציפיות.
• איכות: שיעור שגיאות 4xx/5xx עולה, p99 חביון שובר נורמה היסטורית, עלייה ביחס 404 (סריקה).

השוואת שיטות

אילו שיטות סטטיסטיות משמשות?

שיטות זיהוי סטטיסטיות

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 ימים בהתאם למורכבות המחסנית. קבלו ייעוץ מהנדס כדי להתאים למחסנית שלכם.

איך ליישם את גלאי החריגות: מדריך שלב-אחר-שלב

1. ביקורת תשתית נוכחית. אנו אוספים נתוני מדדים ולוגים היסטוריים למשך 2–4 שבועות, מנתחים פרופיל תעבורה.
2. בחירת אלגוריתם וכיוונון ספים. אנו מגדירים פרמטרי EWMA (alpha 0.05–0.2) ו-z-score (k 3–5) עבור המחסנית שלכם.
3. שילוב עם מקורות מדדים. פריסת הקולטן על Redis/Prometheus, חיבור מצבר לוגים.
4. הגדרת התראות והפחתה אוטומטית. הגדרת ערוצים (Slack, PagerDuty, Telegram) וסקריפטים אוטומטיים להגנה (Cloudflare Under Attack Mode, הגבלת קצב).
5. בדיקה והשקה. אב-טיפוס על שבוע נתונים, בדיקות פורמליות, והפעלה בייצור.

טעויות יישום נפוצות

בעיה שכיחה היא גודל חלון שגוי. חלון קטן מדי גורם להרבה אזעקות שווא, חלון גדול מדי מפספס חריגות מהירות. אנו בוחרים את גודל החלון באופן אישי, תוך ניתוח מחזורים יומיים ושבועיים. טעות אופיינית נוספת היא התעלמות מעונתיות (למשל, Black Friday). הגלאי שלנו משתמש בנתונים היסטוריים מתקופות דומות.

מה כלול בעבודה

• אבחון: ניתוח מדדים ולוגים נוכחיים, בחירת ספים.
• יישום: קוד גלאי (Python, Go, או Lua), הגדרת כללי Prometheus.
• שילוב: Slack, PagerDuty, Telegram, Cloudflare API.
• תיעוד: תיאור אלגוריתם, הוראות פריסה, תרחישי playbook.
• הדרכה: איך להגדיר ולאמן מחדש את הגלאי לשינויי תעבורה.
• תמיכה: אחריות של שבועיים לאחר ייצור (SLA על זמן תגובה).

ציר זמן ועלות

יישום מפתח אורך 3 עד 5 ימי עסקים בהתאם למורכבות התשתית. התמחור נקבע באופן אישי לאחר ביקורת. החיסכון מהיישום יכול להגיע לעשרות אלפי דולרים בשנה על ידי צמצום זמן הזיהוי ב-99% והפחתת זמן השבתה בעד 80%. צרו קשר לייעוץ וקבלו ביקורת של מערכת הניטור שלכם.