AI Log Analysis: Intelligent Error Detection and Root Cause Analysis

from sklearn.ensemble import IsolationForest
import pandas as pd

class LogAnomalyDetector:
    def __init__(self):
        self.model = IsolationForest(contamination=0.01)
        self.vectorizer = TfidfVectorizer(max_features=100)

    def train(self (as discussed in [AI-Powered Security Testing: Finding Vulnerabilities Faster](/blog/ai-security-testing)), normal_logs):
        """Train on normal operating logs"""
        # Vectorize log messages
        log_vectors = self.vectorizer.fit_transform(normal_logs)

        # Train (as discussed in [AI Test Metrics Analytics: Intelligent Analysis of QA Metrics](/blog/ai-test-metrics)) isolation forest
        self.model.fit(log_vectors.toarray())

    def detect_anomalies(self, logs):
        """Detect anomalous log entries"""
        log_vectors = self.vectorizer.transform(logs)
        predictions = self.model.predict(log_vectors.toarray())

        # -1 = anomaly, 1 = normal
        anomalies = [
            {'log': log, 'anomaly_score': score}
            for log, pred, score in zip(logs, predictions, self.model.score_samples(log_vectors.toarray()))
            if pred == -1
        ]

        return anomalies

# Usage
detector = LogAnomalyDetector()
detector.train(historical_normal_logs)

new_logs = fetch_logs(last_hour=True)
anomalies = detector.detect_anomalies(new_logs)

for anomaly in anomalies:
    print(f"Anomaly detected: {anomaly['log']}")
    print(f"Score: {anomaly['anomaly_score']:.3f}")

from sklearn.cluster import DBSCAN
from sentence_transformers import SentenceTransformer

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

    def cluster_logs(self, error_logs):
        """Cluster similar error messages"""
        # Generate embeddings
        embeddings = self.model.encode(error_logs)

        # Cluster with DBSCAN
        clustering = DBSCAN(eps=0.5, min_samples=2).fit(embeddings)

        # Group by cluster
        clusters = {}
        for idx, label in enumerate(clustering.labels_):
            if label not in clusters:
                clusters[label] = []
            clusters[label].append(error_logs[idx])

        return {
            'num_clusters': len(set(clustering.labels_)) - (1 if -1 in clustering.labels_ else 0),
            'clusters': clusters,
            'noise': clusters.get(-1, [])  # Unclustered logs
        }

# Usage
clusterer = LogClusterer()
error_logs = fetch_errors(last_day=True)

result = clusterer.cluster_logs(error_logs)
print(f"Reduced {len(error_logs)} errors to {result['num_clusters']} unique issues")

for cluster_id, logs in result['clusters'].items():
    if cluster_id != -1:  # Skip noise
        print(f"\nCluster {cluster_id} ({len(logs)} occurrences):")
        print(f"Representative: {logs[0]}")

import networkx as nx

class RootCauseAnalyzer:
    def __init__(self):
        self.dependency_graph = nx.DiGraph()

    def build_dependency_graph(self, service_dependencies):
        """Build service dependency graph"""
        for service, deps in service_dependencies.items():
            for dep in deps:
                self.dependency_graph.add_edge(dep, service)

    def analyze_failure(self, failed_service, error_logs, time_window_minutes=5):
        """Trace failure back to root cause"""
        # Get upstream dependencies
        upstream = list(nx.ancestors(self.dependency_graph, failed_service))

        # Analyze errors in time window
        failure_time = error_logs[failed_service]['timestamp']
        root_causes = []

        for upstream_service in upstream:
            if upstream_service in error_logs:
                upstream_error_time = error_logs[upstream_service]['timestamp']

                # Check if upstream failed first
                time_diff = (failure_time - upstream_error_time).total_seconds() / 60

                if 0 < time_diff <= time_window_minutes:
                    root_causes.append({
                        'service': upstream_service,
                        'time_before_failure': time_diff,
                        'error': error_logs[upstream_service]['message']
                    })

        # Sort by time (earliest failure likely root cause)
        root_causes.sort(key=lambda x: x['time_before_failure'], reverse=True)

        return {
            'failed_service': failed_service,
            'likely_root_cause': root_causes[0] if root_causes else None,
            'contributing_factors': root_causes[1:] if len(root_causes) > 1 else []
        }

import re
from transformers import pipeline

class SemanticLogParser:
    def __init__(self):
        self.classifier = pipeline("zero-shot-classification")
        self.categories = [
            "authentication_error",
            "database_timeout",
            "network_failure",
            "memory_error",
            "permission_denied"
        ]

    def parse_log(self, log_message):
        """Extract structured information from log"""
        # Traditional regex for known patterns
        timestamp = re.search(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}', log_message)
        severity = re.search(r'\[(ERROR|WARN|INFO|DEBUG)\]', log_message)

        # AI for semantic categorization
        result = self.classifier(log_message, self.categories)

        return {
            'timestamp': timestamp.group() if timestamp else None,
            'severity': severity.group(1) if severity else 'UNKNOWN',
            'category': result['labels'][0],
            'confidence': result['scores'][0],
            'raw_message': log_message
        }

# Usage
parser = SemanticLogParser()

log = "2025-10-04 14:23:11 [ERROR] Failed to connect to database: timeout after 30s"
parsed = parser.parse_log(log)
print(f"Category: {parsed['category']} (confidence: {parsed['confidence']:.2%})")

from kafka import KafkaConsumer
import json

class RealTimeLogMonitor:
    def __init__(self, anomaly_detector, clusterer):
        self.anomaly_detector = anomaly_detector
        self.clusterer = clusterer
        self.consumer = KafkaConsumer(
            'application-logs',
            bootstrap_servers=['localhost:9092'],
            value_deserializer=lambda m: json.loads(m.decode('utf-8'))
        )

    def monitor(self):
        """Monitor logs in real-time"""
        log_buffer = []

        for message in self.consumer:
            log_entry = message.value

            # Check for anomalies
            if self.anomaly_detector.is_anomalous(log_entry['message']):
                self.send_alert({
                    'severity': 'HIGH',
                    'type': 'anomaly_detected',
                    'log': log_entry
                })

            # Buffer logs for clustering
            if log_entry['level'] == 'ERROR':
                log_buffer.append(log_entry['message'])

            # Periodic clustering (every 100 errors)
            if len(log_buffer) >= 100:
                clusters = self.clusterer.cluster_logs(log_buffer)

                # Alert on new error patterns
                for cluster_id, logs in clusters['clusters'].items():
                    if cluster_id != -1 and len(logs) > 10:
                        self.send_alert({
                            'severity': 'MEDIUM',
                            'type': 'error_spike',
                            'cluster_id': cluster_id,
                            'count': len(logs),
                            'sample': logs[0]
                        })

                log_buffer = []

    def send_alert(self, alert_data):
        """Send alert to incident management system"""
        print(f"ALERT [{alert_data['severity']}]: {alert_data['type']}")
        # Integration with PagerDuty, Slack, etc.

from sklearn.linear_model import LogisticRegression

class FailurePredictor:
    def __init__(self):
        self.model = LogisticRegression()

    def extract_features(self, log_window):
        """Extract features from recent logs"""
        return {
            'error_rate': log_window['level'].value_counts().get('ERROR', 0) / len(log_window),
            'unique_errors': log_window['message'].nunique(),
            'avg_response_time': log_window['response_time_ms'].mean(),
            'p95_response_time': log_window['response_time_ms'].quantile(0.95),
            'database_timeout_count': log_window['message'].str.contains('timeout').sum()
        }

    def predict_failure(self, recent_logs):
        """Predict if system will fail in next N minutes"""
        features = self.extract_features(recent_logs)
        feature_vector = [list(features.values())]

        failure_probability = self.model.predict_proba(feature_vector)[0][1]

        return {
            'failure_probability': failure_probability,
            'alert_threshold': 0.7,
            'should_alert': failure_probability > 0.7,
            'features': features
        }