Análisis de Logs con IA: Detección Inteligente de Errores y Análisis de Causa Raíz

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

class DetectorAnomalíasLog:
    def __init__(self):
        self.modelo = IsolationForest(contamination=0.01)
        self.vectorizador = TfidfVectorizer(max_features=100)

    def entrenar(self, logs_normales):
        """Entrenar en logs de operación normal"""
        vectores_log = self.vectorizador.fit_transform(logs_normales)
        self.modelo.fit(vectores_log.toarray())

    def detectar_anomalías(self, logs):
        """Detectar entradas de log anómalas"""
        vectores_log = self.vectorizador.transform(logs)
        predicciones = self.modelo.predict(vectores_log.toarray())

        anomalías = [
            {'log': log, 'puntaje_anomalía': score}
            for log, pred, score in zip(logs, predicciones, self.modelo.score_samples(vectores_log.toarray()))
            if pred == -1
        ]

        return anomalías

# Uso
detector = DetectorAnomalíasLog()
detector.entrenar(logs_históricos_normales)

nuevos_logs = obtener_logs(última_hora=True)
anomalías = detector.detectar_anomalías(nuevos_logs)

for anomalía in anomalías:
    print(f"Anomalía detectada: {anomalía['log']}")
    print(f"Puntaje: {anomalía['puntaje_anomalía']:.3f}")

from sklearn.cluster import DBSCAN
from sentence_transformers import SentenceTransformer

class ClusterizadorLogs:
    def __init__(self):
        self.modelo = SentenceTransformer('all-MiniLM-L6-v2')

    def clusterizar_logs(self, logs_error):
        """Agrupar mensajes de error similares"""
        embeddings = self.modelo.encode(logs_error)
        clusterización = DBSCAN(eps=0.5, min_samples=2).fit(embeddings)

        clusters = {}
        for idx, etiqueta in enumerate(clusterización.labels_):
            if etiqueta not in clusters:
                clusters[etiqueta] = []
            clusters[etiqueta].append(logs_error[idx])

        return {
            'num_clusters': len(set(clusterización.labels_)) - (1 if -1 in clusterización.labels_ else 0),
            'clusters': clusters,
            'ruido': clusters.get(-1, [])
        }

# Uso
clusterizador = ClusterizadorLogs()
logs_error = obtener_errores(último_día=True)

resultado = clusterizador.clusterizar_logs(logs_error)
print(f"Reducidos {len(logs_error)} errores a {resultado['num_clusters']} problemas únicos")

for cluster_id, logs in resultado['clusters'].items():
    if cluster_id != -1:
        print(f"\nCluster {cluster_id} ({len(logs)} ocurrencias):")
        print(f"Representativo: {logs[0]}")

import networkx as nx

class AnalizadorCausaRaíz:
    def __init__(self):
        self.grafo_dependencias = nx.DiGraph()

    def construir_grafo_dependencias(self, dependencias_servicios):
        """Construir grafo de dependencias de servicios"""
        for servicio, deps in dependencias_servicios.items():
            for dep in deps:
                self.grafo_dependencias.add_edge(dep, servicio)

    def analizar_fallo(self, servicio_fallido, logs_error, ventana_tiempo_minutos=5):
        """Rastrear fallo hasta causa raíz"""
        upstream = list(nx.ancestors(self.grafo_dependencias, servicio_fallido))
        tiempo_fallo = logs_error[servicio_fallido]['timestamp']
        causas_raíz = []

        for servicio_upstream in upstream:
            if servicio_upstream in logs_error:
                tiempo_error_upstream = logs_error[servicio_upstream]['timestamp']
                diferencia_tiempo = (tiempo_fallo - tiempo_error_upstream).total_seconds() / 60

                if 0 < diferencia_tiempo <= ventana_tiempo_minutos:
                    causas_raíz.append({
                        'servicio': servicio_upstream,
                        'tiempo_antes_fallo': diferencia_tiempo,
                        'error': logs_error[servicio_upstream]['mensaje']
                    })

        causas_raíz.sort(key=lambda x: x['tiempo_antes_fallo'], reverse=True)

        return {
            'servicio_fallido': servicio_fallido,
            'causa_raíz_probable': causas_raíz[0] if causas_raíz else None,
            'factores_contribuyentes': causas_raíz[1:] if len(causas_raíz) > 1 else []
        }

import re
from transformers import pipeline

class ParserSemánticoLogs:
    def __init__(self):
        self.clasificador = pipeline("zero-shot-classification")
        self.categorías = [
            "error_autenticación",
            "timeout_base_datos",
            "fallo_red",
            "error_memoria",
            "permiso_denegado"
        ]

    def parsear_log(self, mensaje_log):
        """Extraer información estructurada del log"""
        timestamp = re.search(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}', mensaje_log)
        severidad = re.search(r'\[(ERROR|WARN|INFO|DEBUG)\]', mensaje_log)

        resultado = self.clasificador(mensaje_log, self.categorías)

        return {
            'timestamp': timestamp.group() if timestamp else None,
            'severidad': severidad.group(1) if severidad else 'DESCONOCIDO',
            'categoría': resultado['labels'][0],
            'confianza': resultado['scores'][0],
            'mensaje_crudo': mensaje_log
        }

# Uso
parser = ParserSemánticoLogs()

log = "2025-10-04 14:23:11 [ERROR] Fallo al conectar a base de datos: timeout después de 30s"
parseado = parser.parsear_log(log)
print(f"Categoría: {parseado['categoría']} (confianza: {parseado['confianza']:.2%})")

from kafka import KafkaConsumer
import json

class MonitorLogsReealTiempo:
    def __init__(self, detector_anomalías, clusterizador):
        self.detector_anomalías = detector_anomalías
        self.clusterizador = clusterizador
        self.consumidor = KafkaConsumer(
            'application-logs',
            bootstrap_servers=['localhost:9092'],
            value_deserializer=lambda m: json.loads(m.decode('utf-8'))
        )

    def monitorear(self):
        """Monitorear logs en tiempo real"""
        buffer_logs = []

        for mensaje in self.consumidor:
            entrada_log = mensaje.value

            if self.detector_anomalías.es_anómalo(entrada_log['mensaje']):
                self.enviar_alerta({
                    'severidad': 'ALTA',
                    'tipo': 'anomalía_detectada',
                    'log': entrada_log
                })

            if entrada_log['nivel'] == 'ERROR':
                buffer_logs.append(entrada_log['mensaje'])

            if len(buffer_logs) >= 100:
                clusters = self.clusterizador.clusterizar_logs(buffer_logs)

                for cluster_id, logs in clusters['clusters'].items():
                    if cluster_id != -1 and len(logs) > 10:
                        self.enviar_alerta({
                            'severidad': 'MEDIA',
                            'tipo': 'pico_errores',
                            'cluster_id': cluster_id,
                            'conteo': len(logs),
                            'muestra': logs[0]
                        })

                buffer_logs = []

    def enviar_alerta(self, datos_alerta):
        """Enviar alerta al sistema de gestión de incidentes"""
        print(f"ALERTA [{datos_alerta['severidad']}]: {datos_alerta['tipo']}")
        # Integración con PagerDuty, Slack, etc.

from sklearn.linear_model import LogisticRegression

class PredictorFallos:
    def __init__(self):
        self.modelo = LogisticRegression()

    def extraer_características(self, ventana_log):
        """Extraer características de logs recientes"""
        return {
            'tasa_error': ventana_log['nivel'].value_counts().get('ERROR', 0) / len(ventana_log),
            'errores_únicos': ventana_log['mensaje'].nunique(),
            'tiempo_respuesta_promedio': ventana_log['tiempo_respuesta_ms'].mean(),
            'tiempo_respuesta_p95': ventana_log['tiempo_respuesta_ms'].quantile(0.95),
            'conteo_timeout_bd': ventana_log['mensaje'].str.contains('timeout').sum()
        }

    def predecir_fallo(self, logs_recientes):
        """Predecir si el sistema fallará en los próximos N minutos"""
        características = self.extraer_características(logs_recientes)
        vector_características = [list(características.values())]

        probabilidad_fallo = self.modelo.predict_proba(vector_características)[0][1]

        return {
            'probabilidad_fallo': probabilidad_fallo,
            'umbral_alerta': 0.7,
            'debe_alertar': probabilidad_fallo > 0.7,
            'características': características
        }