ИИ Анализ Логов: Интеллектуальное Обнаружение Ошибок и Анализ Первопричин

from sklearn.ensemble import IsolationForest
from sklearn.feature_extraction.text import TfidfVectorizer
import pandas as pd

class ДетекторАномалийЛогов:
    def __init__(self):
        self.модель = IsolationForest(contamination=0.01)
        self.векторизатор = TfidfVectorizer(max_features=100)

    def обучить(self, нормальные_логи):
        """Обучить на нормальных операционных логах"""
        векторы_логов = self.векторизатор.fit_transform(нормальные_логи)
        self.модель.fit(векторы_логов.toarray())

    def обнаружить_аномалии(self, логи):
        """Обнаружить аномальные записи логов"""
        векторы_логов = self.векторизатор.transform(логи)
        предсказания = self.модель.predict(векторы_логов.toarray())

        аномалии = [
            {'лог': лог, 'оценка_аномалии': оценка}
            for лог, пред, оценка in zip(логи, предсказания, self.модель.score_samples(векторы_логов.toarray()))
            if пред == -1
        ]

        return аномалии

# Использование
детектор = ДетекторАномалийЛогов()
детектор.обучить(исторические_нормальные_логи)

новые_логи = получить_логи(последний_час=True)
аномалии = детектор.обнаружить_аномалии(новые_логи)

for аномалия in аномалии:
    print(f"Обнаружена аномалия: {аномалия['лог']}")
    print(f"Оценка: {аномалия['оценка_аномалии']:.3f}")

from sklearn.cluster import DBSCAN
from sentence_transformers import SentenceTransformer

class КластеризаторЛогов:
    def __init__(self):
        self.модель = SentenceTransformer('all-MiniLM-L6-v2')

    def кластеризовать_логи(self, логи_ошибок):
        """Группировать похожие сообщения об ошибках"""
        эмбеддинги = self.модель.encode(логи_ошибок)
        кластеризация = DBSCAN(eps=0.5, min_samples=2).fit(эмбеддинги)

        кластеры = {}
        for индекс, метка in enumerate(кластеризация.labels_):
            if метка not in кластеры:
                кластеры[метка] = []
            кластеры[метка].append(логи_ошибок[индекс])

        return {
            'количество_кластеров': len(set(кластеризация.labels_)) - (1 if -1 in кластеризация.labels_ else 0),
            'кластеры': кластеры,
            'шум': кластеры.get(-1, [])
        }

# Использование
кластеризатор = КластеризаторЛогов()
логи_ошибок = получить_ошибки(последний_день=True)

результат = кластеризатор.кластеризовать_логи(логи_ошибок)
print(f"Сокращено {len(логи_ошибок)} ошибок до {результат['количество_кластеров']} уникальных проблем")

for ид_кластера, логи in результат['кластеры'].items():
    if ид_кластера != -1:
        print(f"\nКластер {ид_кластера} ({len(логи)} вхождений):")
        print(f"Представитель: {логи[0]}")

import networkx as nx

class АнализаторПервопричин:
    def __init__(self):
        self.граф_зависимостей = nx.DiGraph()

    def построить_граф_зависимостей(self, зависимости_сервисов):
        """Построить граф зависимостей сервисов"""
        for сервис, зависимости in зависимости_сервисов.items():
            for зависимость in зависимости:
                self.граф_зависимостей.add_edge(зависимость, сервис)

    def анализировать_сбой(self, упавший_сервис, логи_ошибок, окно_времени_минуты=5):
        """Проследить сбой до первопричины"""
        вышестоящие = list(nx.ancestors(self.граф_зависимостей, упавший_сервис))
        время_сбоя = логи_ошибок[упавший_сервис]['timestamp']
        первопричины = []

        for вышестоящий_сервис in вышестоящие:
            if вышестоящий_сервис in логи_ошибок:
                время_ошибки_вышестоящего = логи_ошибок[вышестоящий_сервис]['timestamp']
                разница_времени = (время_сбоя - время_ошибки_вышестоящего).total_seconds() / 60

                if 0 < разница_времени <= окно_времени_минуты:
                    первопричины.append({
                        'сервис': вышестоящий_сервис,
                        'время_до_сбоя': разница_времени,
                        'ошибка': логи_ошибок[вышестоящий_сервис]['сообщение']
                    })

        первопричины.sort(key=lambda x: x['время_до_сбоя'], reverse=True)

        return {
            'упавший_сервис': упавший_сервис,
            'вероятная_первопричина': первопричины[0] if первопричины else None,
            'способствующие_факторы': первопричины[1:] if len(первопричины) > 1 else []
        }

import re
from transformers import pipeline

class СемантическийПарсерЛогов:
    def __init__(self):
        self.классификатор = pipeline("zero-shot-classification")
        self.категории = [
            "ошибка_аутентификации",
            "таймаут_базы_данных",
            "сбой_сети",
            "ошибка_памяти",
            "отказ_доступа"
        ]

    def распарсить_лог(self, сообщение_лога):
        """Извлечь структурированную информацию из лога"""
        timestamp = re.search(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}', сообщение_лога)
        серьёзность = re.search(r'\[(ERROR|WARN|INFO|DEBUG)\]', сообщение_лога)

        результат = self.классификатор(сообщение_лога, self.категории)

        return {
            'timestamp': timestamp.group() if timestamp else None,
            'серьёзность': серьёзность.group(1) if серьёзность else 'НЕИЗВЕСТНО',
            'категория': результат['labels'][0],
            'уверенность': результат['scores'][0],
            'сырое_сообщение': сообщение_лога
        }

# Использование
парсер = СемантическийПарсерЛогов()

лог = "2025-10-04 14:23:11 [ERROR] Не удалось подключиться к базе данных: таймаут после 30s"
распарсено = парсер.распарсить_лог(лог)
print(f"Категория: {распарсено['категория']} (уверенность: {распарсено['уверенность']:.2%})")

from kafka import KafkaConsumer
import json

class МониторЛоговРеальногоВремени:
    def __init__(self, детектор_аномалий, кластеризатор):
        self.детектор_аномалий = детектор_аномалий
        self.кластеризатор = кластеризатор
        self.потребитель = KafkaConsumer(
            'application-logs',
            bootstrap_servers=['localhost:9092'],
            value_deserializer=lambda m: json.loads(m.decode('utf-8'))
        )

    def мониторить(self):
        """Мониторить логи в реальном времени"""
        буфер_логов = []

        for сообщение in self.потребитель:
            запись_лога = сообщение.value

            if self.детектор_аномалий.аномальный(запись_лога['сообщение']):
                self.отправить_алерт({
                    'серьёзность': 'ВЫСОКАЯ',
                    'тип': 'аномалия_обнаружена',
                    'лог': запись_лога
                })

            if запись_лога['уровень'] == 'ERROR':
                буфер_логов.append(запись_лога['сообщение'])

            if len(буфер_логов) >= 100:
                кластеры = self.кластеризатор.кластеризовать_логи(буфер_логов)

                for ид_кластера, логи in кластеры['кластеры'].items():
                    if ид_кластера != -1 and len(логи) > 10:
                        self.отправить_алерт({
                            'серьёзность': 'СРЕДНЯЯ',
                            'тип': 'всплеск_ошибок',
                            'ид_кластера': ид_кластера,
                            'количество': len(логи),
                            'образец': логи[0]
                        })

                буфер_логов = []

    def отправить_алерт(self, данные_алерта):
        """Отправить алерт в систему управления инцидентами"""
        print(f"АЛЕРТ [{данные_алерта['серьёзность']}]: {данные_алерта['тип']}")
        # Интеграция с PagerDuty, Slack и т.д.

from sklearn.linear_model import LogisticRegression

class ПредикторСбоев:
    def __init__(self):
        self.модель = LogisticRegression()

    def извлечь_характеристики(self, окно_логов):
        """Извлечь характеристики из недавних логов"""
        return {
            'частота_ошибок': окно_логов['уровень'].value_counts().get('ERROR', 0) / len(окно_логов),
            'уникальные_ошибки': окно_логов['сообщение'].nunique(),
            'среднее_время_ответа': окно_логов['время_ответа_мс'].mean(),
            'время_ответа_p95': окно_логов['время_ответа_мс'].quantile(0.95),
            'количество_таймаутов_бд': окно_логов['сообщение'].str.contains('таймаут').sum()
        }

    def предсказать_сбой(self, недавние_логи):
        """Предсказать произойдет ли сбой системы в ближайшие N минут"""
        характеристики = self.извлечь_характеристики(недавние_логи)
        вектор_характеристик = [list(характеристики.values())]

        вероятность_сбоя = self.модель.predict_proba(вектор_характеристик)[0][1]

        return {
            'вероятность_сбоя': вероятность_сбоя,
            'порог_алерта': 0.7,
            'должен_алертить': вероятность_сбоя > 0.7,
            'характеристики': характеристики
        }