# Fluent Bit collection configuration
[INPUT]
    Name              tail
    Path              /var/log/nginx/*.log
    Tag               nginx.access
    Parser            nginx
    DB                /var/lib/fluent-bit/nginx.db
    Refresh_Interval  5

[INPUT]
    Name              tail
    Path              /var/log/syslog
    Tag               system.syslog
    DB                /var/lib/fluent-bit/syslog.db

[FILTER]
    Name                lua
    Match               system.*
    Script              /fluent-bit/scripts/classify.lua
    Call                classify_log_level

[OUTPUT]
    Name                http
    Match               *
    Host                log-processor
    Port                8080
    Format              json

import numpy as np
from sklearn.cluster import DBSCAN
from sentence_transformers import SentenceTransformer
import hashlib

class LogPatternClusterer:
    """LLM embedding-based log pattern clusterer"""
    
    def __init__(self, model_name="text-embedding-3-small"):
        self.embedder = SentenceTransformer(model_name)
        self.cluster_threshold = 0.15  # Similarity threshold
        self.pattern_registry = {}  # pattern_hash -> pattern_info
    
    def extract_log_template(self, log_line: str) -> str:
        """Extract template from log line (remove dynamic fields)"""
        # LLM-based template extraction
        # Example: "2024-01-15 10:23:45 ERROR Connection to db-01 failed: timeout after 30s"
        # Template: "Connection to {host} failed: timeout after {seconds}s"
        import re
        # Remove timestamps
        template = re.sub(r'\d{4}-\d{2}-\d{2}[\sT]\d{2}:\d{2}:\d{2}', '{timestamp}', log_line)
        # Remove IP addresses
        template = re.sub(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', '{ip}', template)
        # Remove numbers
        template = re.sub(r'\b\d+\b', '{num}', template)
        # Remove UUIDs
        template = re.sub(r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', '{uuid}', template)
        return template
    
    def cluster_logs(self, log_batch: list[dict]) -> dict:
        """Cluster a batch of logs"""
        if not log_batch:
            return {}
        
        # Extract templates
        templates = [self.extract_log_template(log['message']) for log in log_batch]
        
        # Generate embeddings
        embeddings = self.embedder.encode(templates)
        
        # DBSCAN clustering
        clusterer = DBSCAN(eps=self.cluster_threshold, min_samples=1, metric='cosine')
        labels = clusterer.fit_predict(embeddings)
        
        # Build cluster results
        clusters = {}
        for log, label in zip(log_batch, labels):
            cluster_key = f"cluster_{label}"
            if cluster_key not in clusters:
                clusters[cluster_key] = {
                    'pattern': templates[log_batch.index(log)],
                    'count': 0,
                    'samples': [],
                    'first_seen': log['timestamp'],
                    'last_seen': log['timestamp'],
                    'hosts': set(),
                    'error_rate': 0.0
                }
            clusters[cluster_key]['count'] += 1
            clusters[cluster_key]['samples'].append(log['message'])
            clusters[cluster_key]['last_seen'] = log['timestamp']
            clusters[cluster_key]['hosts'].add(log.get('host', 'unknown'))
            
            if 'error' in log['message'].lower() or 'fail' in log['message'].lower():
                clusters[cluster_key]['error_rate'] = 1.0
        
        return clusters

-- Log vector storage table
CREATE TABLE log_patterns (
    id BIGSERIAL PRIMARY KEY,
    pattern_hash VARCHAR(64) UNIQUE NOT NULL,
    pattern_template TEXT NOT NULL,
    embedding vector(1536),  -- OpenAI text-embedding-3-small
    total_count BIGINT DEFAULT 0,
    error_count BIGINT DEFAULT 0,
    first_seen TIMESTAMP WITH TIME ZONE,
    last_seen TIMESTAMP WITH TIME ZONE,
    related_hosts TEXT[],
    severity_score DOUBLE PRECISION,
    trend_direction VARCHAR(10),  -- 'increasing', 'stable', 'decreasing'
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

-- Create vector index for accelerated similarity search
CREATE INDEX ON log_patterns USING ivfflat (embedding vector_cosine_ops)
    WITH (lists = 100);

-- Trend prediction table
CREATE TABLE log_trend_predictions (
    id BIGSERIAL PRIMARY KEY,
    pattern_hash VARCHAR(64) REFERENCES log_patterns(pattern_hash),
    predicted_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    predicted_count_1h DOUBLE PRECISION,
    predicted_count_24h DOUBLE PRECISION,
    confidence_score DOUBLE PRECISION,
    prediction_model VARCHAR(50),
    actual_1h_count BIGINT,
    actual_24h_count BIGINT
);

class HierarchicalLogClustering:
    """Hierarchical log clustering: coarse template matching, fine semantic clustering"""
    
    def __init__(self, llm_client, embedding_model):
        self.llm = llm_client
        self.embedder = embedding_model
        self.template_index = {}  # template -> cluster_id
        self.semantic_clusters = {}  # semantic cluster -> log list
    
    def process_log_batch(self, logs: list[dict]) -> list[dict]:
        """Process log batch, return logs with cluster labels"""
        # Layer 1: Template matching (fast filtering)
        template_clusters = self._template_match(logs)
        
        # Layer 2: Semantic clustering (LLM analysis for unmatched logs)
        unmatched = [log for log in logs if log.get('cluster_id') is None]
        semantic_clusters = self._semantic_cluster(unmatched)
        
        # Layer 3: LLM semantic enrichment (LLM understanding for key clusters)
        enhanced_clusters = self._llm_enrich(template_clusters, semantic_clusters)
        
        return enhanced_clusters
    
    def _llm_enrich(self, template_clusters, semantic_clusters) -> dict:
        """Enrich cluster results with LLM semantic understanding"""
        enrichment_prompt = """
Analyze the following log clusters and identify their common patterns and potential root causes:

Cluster List:
{clusters}

For each cluster, provide:
1. Human-readable pattern description
2. Potential root cause analysis
3. Suggested remediation direction
4. Urgency assessment (P0-P3)

Return analysis results in JSON format.
""".format(clusters=json.dumps(template_clusters, ensure_ascii=False, indent=2))
        
        response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": enrichment_prompt}],
            response_format={"type": "json_object"}
        )
        return json.loads(response.choices[0].message.content)

from asyncio import Queue, create_task
import asyncio

class RealtimeLogPipeline:
    """Real-time log processing pipeline"""
    
    def __init__(self, clusterer: LogPatternClusterer, 
                 detector: AnomalyDetector,
                 predictor: TrendPredictor):
        self.queue = Queue(maxsize=10000)
        self.clusterer = clusterer
        self.detector = detector
        self.predictor = predictor
        self.batch_size = 50
        self.processing_tasks = []
    
    async def start(self):
        """Start the processing pipeline"""
        create_task(self._batch_processor())
        create_task(self._alert_processor())
        create_task(self._prediction_processor())
    
    async def _batch_processor(self):
        """Batch process logs, perform clustering"""
        batch = []
        while True:
            log = await asyncio.wait_for(self.queue.get(), timeout=1.0)
            batch.append(log)
            
            if len(batch) >= self.batch_size:
                clusters = self.clusterer.cluster_logs(batch)
                await self._store_clusters(clusters)
                batch = []
            
            # Force process on timeout
            if len(batch) > 0:
                await asyncio.sleep(0.5)
                clusters = self.clusterer.cluster_logs(batch)
                await self._store_clusters(clusters)
                batch = []
    
    async def ingest(self, log: dict):
        """Ingest log into pipeline"""
        await self.queue.put(log)

import statsmodels.api as sm
from prophet import Prophet
import pandas as pd

class LogAnomalyDetector:
    """Time-series based log anomaly detection"""
    
    def __init__(self, lookback_hours=24, confidence=0.95):
        self.lookback_hours = lookback_hours
        self.confidence = confidence
        self.baselines = {}  # pattern -> baseline stats
    
    def update_baseline(self, pattern: str, counts: list[tuple]):
        """Update baseline for a log pattern"""
        df = pd.DataFrame(counts, columns=['ds', 'y'])
        
        # Decompose time series
        decomposition = sm.tsa.seasonal_decompose(
            df['y'], model='additive', period=24
        )
        
        self.baselines[pattern] = {
            'trend': decomposition.trend,
            'seasonal': decomposition.seasonal,
            'residual': decomposition.residual,
            'mean': df['y'].mean(),
            'std': df['y'].std(),
            'upper_bound': df['y'].mean() + 2 * df['y'].std(),
            'lower_bound': max(0, df['y'].mean() - 2 * df['y'].std())
        }
    
    def detect_anomaly(self, pattern: str, current_count: int) -> dict:
        """Detect if current count is anomalous"""
        if pattern not in self.baselines:
            return {'is_anomaly': False, 'reason': 'no_baseline'}
        
        baseline = self.baselines[pattern]
        z_score = (current_count - baseline['mean']) / baseline['std'] if baseline['std'] > 0 else 0
        is_anomaly = abs(z_score) > 2  # 2σ threshold
        
        prophet_anomaly = self._prophet_detect(pattern, current_count)
        
        return {
            'is_anomaly': is_anomaly or prophet_anomaly['is_anomaly'],
            'z_score': z_score,
            'expected_range': (baseline['lower_bound'], baseline['upper_bound']),
            'current_value': current_count,
            'prophet_anomaly': prophet_anomaly,
            'severity': self._calculate_severity(z_score, current_count, baseline['mean'])
        }
    
    def _calculate_severity(self, z_score: float, current: int, mean: int) -> str:
        """Calculate anomaly severity"""
        if abs(z_score) > 4:
            return 'critical'
        elif abs(z_score) > 3:
            return 'high'
        elif abs(z_score) > 2:
            return 'medium'
        return 'low'

class LogTrendPredictor:
    """Log trend prediction"""
    
    def __init__(self, forecast_horizon_hours=24):
        self.forecast_horizon = forecast_horizon_hours
        self.models = {}
    
    def fit(self, pattern: str, history: list[dict]):
        """Train prediction model"""
        df = pd.DataFrame(history)
        df['ds'] = pd.to_datetime(df['timestamp'])
        df['y'] = df['count']
        
        model = Prophet(
            yearly_seasonality=True,
            weekly_seasonality=True,
            daily_seasonality=True,
            changepoint_prior_scale=0.05
        )
        model.fit(df)
        self.models[pattern] = model
    
    def predict(self, pattern: str, steps: int = 96) -> dict:
        """Predict future trends (every 15 min for 24 hours)"""
        if pattern not in self.models:
            return {'forecast': [], 'confidence': []}
        
        model = self.models[pattern]
        future = model.make_future_dataframe(periods=steps, freq='15min')
        forecast = model.predict(future)
        
        recent_forecast = forecast.tail(steps)
        
        return {
            'forecast': recent_forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].values.tolist(),
            'trend_direction': self._infer_trend(recent_forecast['yhat'].values),
            'peak_predicted': float(recent_forecast['yhat'].max()),
            'peak_time': str(recent_forecast.loc[recent_forecast['yhat'].idxmax(), 'ds'])
        }
    
    def _infer_trend(self, values: np.ndarray) -> str:
        """Infer trend direction"""
        if len(values) < 2:
            return 'stable'
        slope = np.polyfit(range(len(values)), values, 1)[0]
        if slope > 0.01 * abs(values.mean()):
            return 'increasing'
        elif slope < -0.01 * abs(values.mean()):
            return 'decreasing'
        return 'stable'

# docker-compose.yml
version: '3.8'

services:
  # Log collection
  fluent-bit:
    image: fluent/fluent-bit:3.1
    volumes:
      - ./fluent-bit/conf:/fluent-bit/etc
      - /var/log:/var/log:ro
      - /var/lib/docker/containers:/var/lib/docker/containers:ro
    networks:
      - lognet

  # AI log analysis engine
  log-processor:
    build: ./log-processor
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - EMBEDDING_MODEL=text-embedding-3-small
      - DATABASE_URL=postgresql://loguser:logpass@postgres:5432/logdb
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - postgres
      - redis
    networks:
      - lognet
    deploy:
      resources:
        limits:
          memory: 2G
          cpus: '1.0'

  # Vector database
  postgres:
    image: pgvector/pgvector:pg16
    environment:
      POSTGRES_DB: logdb
      POSTGRES_USER: loguser
      POSTGRES_PASSWORD: logpass
    volumes:
      - postgres_data:/var/lib/postgresql/data
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
    networks:
      - lognet

  # Cache layer
  redis:
    image: redis:7-alpine
    command: redis-server --appendonly yes
    volumes:
      - redis_data:/data
    networks:
      - lognet

  # Visualization
  grafana:
    image: grafana/grafana:11.3.0
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD}
      - GF_INSTALL_PLUGINS=grafana-clock-panel
    volumes:
      - grafana_data:/var/lib/grafana
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards
      - ./grafana/datasources:/etc/grafana/provisioning/datasources
    networks:
      - lognet

  # Alert management
  alertmanager:
    image: prom/alertmanager:v0.27.0
    volumes:
      - ./alertmanager:/etc/alertmanager
    networks:
      - lognet

volumes:
  postgres_data:
  redis_data:
  grafana_data:

networks:
  lognet:
    driver: bridge

# alert-rules.yaml
groups:
  - name: log_anomaly_alerts
    rules:
      - alert: LogPatternAnomaly
        expr: log_anomaly_score > 3
        for: 5m
        labels:
          severity: '{{ $value | lessThan 4 | ternary "warning" | ternary "critical" }}'
          category: 'log-anomaly'
        annotations:
          summary: "Log pattern anomaly: {{ $labels.pattern }}"
          description: "Pattern {{ $labels.pattern }} deviates {{ $value }}σ in current period, immediate review recommended"
          runbook_url: "https://wiki.example.com/runbooks/log-anomaly"

      - alert: LogVolumeSpike
        expr: log_volume_prediction_deviation > 2
        for: 10m
        labels:
          severity: 'warning'
          category: 'log-volume'
        annotations:
          summary: "Log volume spike predicted"
          description: "Log volume prediction deviation exceeds 2σ, potential issue may exist"

      - alert: NewLogPatternDetected
        expr: new_pattern_count_1h > 5
        for: 15m
        labels:
          severity: 'info'
          category: 'new-pattern'
        annotations:
          summary: "New log patterns detected"
          description: "{{ $value }} new log patterns detected in the past hour, may require analysis"

  - name: log_trend_alerts
    rules:
      - alert: ErrorRateIncreasing
        expr: error_rate_trend == "increasing" and error_rate_7d_avg > 0.05
        for: 30m
        labels:
          severity: 'warning'
          category: 'error-trend'
        annotations:
          summary: "Error rate showing increasing trend"
          description: "Error rate has been trending upward over 7 days, current 7-day average: {{ $value | humanizePercentage }}"