Разработка AI-системы мониторинга инфраструктуры и автоматического алертинга (AIOps)

Разработка AI-системы мониторинга инфраструктуры и автоматического алертинга (AIOps)

AIOps (Artificial Intelligence for IT Operations) объединяет machine learning с данными мониторинга — метриками, логами, трассировками — для снижения MTTR, шумоподавления и предиктивной диагностики. Для команды из 5-10 инженеров, управляющей сотнями сервисов, AIOps — единственный способ оставаться эффективными.

Проблемы классического мониторинга

Alert storm: Один инцидент генерирует сотни алертов из взаимосвязанных систем. Инженер тонет в уведомлениях и пропускает корневую причину.

Static thresholds: Порог CPU > 80% — ложная тревога ночью при batch job и пропуск проблемы днём при нормальных 75% + тренд роста.

Ручная корреляция: Инженер вручную сопоставляет метрики, логи, трассировки из разных систем. AIOps автоматизирует это.

Инфраструктура данных

Unified Observability Pipeline:

Application Metrics → Prometheus
Infrastructure Metrics → Node Exporter, cloud_exporter
Logs → Fluent Bit → Elasticsearch / Loki
Traces → OpenTelemetry → Jaeger / Tempo
Events → Kubernetes Events
    ↓
Kafka (общий event bus)
    ↓
AIOps Platform (ML inference + correlation)
    ↓
Alert Manager → PagerDuty / OpsGenie / Slack

Объём данных: Enterprise: миллиарды метрических точек в сутки. Требования к ML pipeline: batch для обучения, streaming для inference < 30 секунд latency.

Динамические пороги

Prophets/SARIMA для seasonal thresholds:

from prophet import Prophet
import pandas as pd

def train_dynamic_threshold(metric_series, confidence_level=0.99):
    """
    Prophet обучается на нормальном поведении метрики
    Возвращает upper/lower band для каждого момента времени
    """
    df = pd.DataFrame({
        'ds': metric_series.index,
        'y': metric_series.values
    })

    model = Prophet(
        seasonality_mode='multiplicative',
        weekly_seasonality=True,
        daily_seasonality=True,
        interval_width=confidence_level
    )
    model.fit(df)

    future = model.make_future_dataframe(periods=60, freq='5min')
    forecast = model.predict(future)

    # Аномалия: значение вне yhat_lower / yhat_upper
    return forecast[['ds', 'yhat_lower', 'yhat_upper']]

def detect_anomaly(current_value, forecast_row):
    return (current_value < forecast_row['yhat_lower'] or
            current_value > forecast_row['yhat_upper'])

EWMA для real-time адаптации:

class EWMAThreshold:
    def __init__(self, alpha=0.1, k=3.0):
        self.alpha = alpha  # скорость адаптации
        self.k = k          # количество сигм
        self.ewma = None
        self.ewmv = None    # EWMA variance

    def update(self, value):
        if self.ewma is None:
            self.ewma = value
            self.ewmv = 0
            return False

        deviation = value - self.ewma
        self.ewma = self.alpha * value + (1 - self.alpha) * self.ewma
        self.ewmv = self.alpha * deviation**2 + (1 - self.alpha) * self.ewmv

        threshold = self.k * np.sqrt(self.ewmv)
        is_anomaly = abs(deviation) > threshold

        return is_anomaly

Корреляция алертов и шумоподавление

Alert Clustering:

from sklearn.cluster import DBSCAN
import numpy as np

def cluster_alerts(alerts_df, temporal_eps=300, spatial_eps=0.5):
    """
    Кластеризация алертов: временная близость + семантическая схожесть
    """
    # Фичи: timestamp + service embedding + severity
    features = np.column_stack([
        alerts_df['timestamp'].astype(int) / 1e9,    # Unix timestamp
        alerts_df['service_embedding'],               # Word2Vec/FastText
        alerts_df['severity_numeric']
    ])

    # Нормализация
    from sklearn.preprocessing import StandardScaler
    features_scaled = StandardScaler().fit_transform(features)

    clusters = DBSCAN(eps=0.5, min_samples=2).fit_predict(features_scaled)
    alerts_df['incident_cluster'] = clusters

    return alerts_df.groupby('incident_cluster').agg({
        'alert_id': 'count',
        'service': lambda x: x.mode()[0],
        'severity': 'max',
        'timestamp': 'min',
        'message': list
    })

Causal Graph для RCA:

import networkx as nx

class ServiceDependencyGraph:
    def __init__(self):
        self.graph = nx.DiGraph()

    def build_from_traces(self, traces):
        for trace in traces:
            for span in trace.spans:
                if span.parent:
                    self.graph.add_edge(span.parent_service, span.service,
                                        latency=span.latency)

    def find_root_cause(self, incident_services, anomaly_time):
        """
        Из списка аномальных сервисов → найти upstream корень
        """
        anomaly_set = set(incident_services)
        root_candidates = []

        for service in anomaly_set:
            # Предки: если предок тоже аномальный → он ближе к корню
            ancestors = nx.ancestors(self.graph, service)
            if not ancestors.intersection(anomaly_set):
                root_candidates.append(service)

        return root_candidates

Предиктивная диагностика

Тренды до инцидента:

def detect_precursor_patterns(metrics_history, incident_labels, window=2*60):
    """
    Обучение: за сколько минут до инцидента появляются первые признаки?
    """
    precursor_features = []
    for incident_time in incident_labels:
        # Окно за 2 часа до инцидента
        pre_incident = metrics_history[
            incident_time - timedelta(minutes=window):incident_time
        ]

        features = {
            'error_rate_trend': np.polyfit(range(window), pre_incident['error_rate'], 1)[0],
            'p99_latency_trend': np.polyfit(range(window), pre_incident['p99'], 1)[0],
            'cpu_trend': np.polyfit(range(window), pre_incident['cpu'], 1)[0],
            'memory_leak_indicator': detect_memory_leak_pattern(pre_incident['memory'])
        }
        precursor_features.append(features)

    # Модель: за 30 минут до инцидента → предсказание
    incident_predictor = LogisticRegression()
    incident_predictor.fit(precursor_features, labels)
    return incident_predictor

LLM-assisted incident analysis

Контекстный анализ инцидента:

from anthropic import Anthropic

def generate_incident_summary(incident_data, llm_client):
    """
    LLM генерирует читаемое резюме инцидента из технических данных
    """
    context = f"""
    Incident at {incident_data['timestamp']}:
    - Affected services: {incident_data['services']}
    - Anomalous metrics: {incident_data['metrics']}
    - Recent changes: {incident_data['recent_deployments']}
    - Related logs (sample): {incident_data['log_samples'][:5]}
    - Similar past incidents: {incident_data['similar_incidents']}
    """

    response = llm_client.messages.create(
        model='claude-opus-4',
        messages=[{
            'role': 'user',
            'content': f'Provide a concise incident analysis and recommended next steps:\n{context}'
        }]
    )

    return response.content[0].text

Интеграция

Grafana AIOps Plugin: Единый дашборд: аномалии, кластеры алертов, RCA граф, precursor warnings.

PagerDuty Event Intelligence: Коммерческая AIOps надстройка. Кастомный ML можно интегрировать через PagerDuty Events API.

ServiceNow AIOps: Автоматическое создание инцидентов с ML-классификацией priority и assignment group.

Стек: Prometheus + Grafana (метрики) + Loki (логи) + Tempo (трассировки) + Kafka (stream) + ClickHouse (analytics) + FastAPI (ML inference) + React (custom AIOps UI).

Сроки: динамические пороги + alert clustering + Slack интеграция — 4-5 недель. Causal graph RCA, precursor detection, LLM incident summary, PagerDuty/ServiceNow integration — 3-4 месяца.