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

from sklearn.ensemble import IsolationForest

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

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, [])
        }

import networkx as nx

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

    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 []
        }

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"
        ]

    def parsear_log(self, mensaje_log):
        """Extraer información estructurada del log"""
        resultado = self.clasificador(mensaje_log, self.categorías)

        return {
            'categoría': resultado['labels'][0],
            'confianza': resultado['scores'][0],
            'mensaje_crudo': mensaje_log
        }

class MonitorLogsReealTiempo:
    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)
                        })

                buffer_logs = []