log_format ai_monitor '$remote_addr - $request_time $upstream_response_time '
                      '$upstream_status $request_length $bytes_sent '
                      '$http_referer $http_user_agent '
                      '$http_x_forwarded_for $http_x_real_ip '
                      '$host $server_port $request_method '
                      '$upstream_addr $status $body_bytes_sent';

# vector.toml
sources:
  nginx_logs:
    type: file
    include: ["/var/log/nginx/access.log"]
    read_from: beginning
    
transforms:
  parse_nginx:
    type: remap
    source: |
      . = parse_nginx_log!(.message)
      
sinks:
  prometheus:
    type: prometheus_exporter
    inputs: ["parse_nginx"]
    
  loki:
    type: loki
    inputs: ["parse_nginx"]
    endpoint: "http://loki:3100"
    labels:
      service: nginx
      environment: production

# prometheus_rules.yml
groups:
  - name: ai_traffic_rules
    rules:
      - record: ai:request_latency_p99
        expr: histogram_quantile(0.99, sum(rate(nginx_request_duration_seconds_bucket[5m])) by (le, upstream))
        
      - record: ai:error_rate_by_upstream
        expr: sum(rate(nginx_http_requests_total{status=~"5.."}[5m])) by (upstream) / sum(rate(nginx_http_requests_total[5m])) by (upstream)
        
      - record: ai:active_connections_per_upstream
        expr: nginx_connections_active / count by (instance) (nginx_connections_active)
        
      - record: ai:geo_latency
        expr: histogram_quantile(0.5, sum by (le, geo_zone) (rate(nginx_request_duration_seconds_bucket[5m])))

import pandas as pd
import numpy as np
from prophet import Prophet
from sklearn.ensemble import RandomForestRegressor
import prometheus_api_client

class TrafficPredictor:
    """基于历史数据的流量预测器"""
    
    def __init__(self, lookback_days=14):
        self.lookback_days = lookback_days
        self.models = {}
        self.feature_names = [
            'hour_of_day', 'day_of_week', 'is_weekend',
            'avg_latency_last_hour', 'error_rate_last_hour',
            'bandwidth_mb_last_hour'
        ]
    
    def fetch_metrics(self, prometheus):
        """从 Prometheus 采集历史数据"""
        query = """
        avg_over_time(nginx_http_requests_total[1h])
        by (upstream, method, path)
        """
        result = prometheus.CustomMetrics.get_metrics(query)
        return result
    
    def train_time_series(self, upstream):
        """为每个后端训练时间序列预测模型"""
        df = self._get_history_df(upstream)
        
        model = Prophet(
            yearly_seasonality=True,
            weekly_seasonality=True,
            daily_seasonality=True,
            changepoint_prior_scale=0.05
        )
        model.fit(df)
        self.models[upstream] = model
        return model
    
    def predict_next_hour(self, upstream, hours_ahead=4):
        """预测未来 N 小时的流量"""
        if upstream not in self.models:
            self.train_time_series(upstream)
        
        future = self.models[upstream].make_future_dataframe(periods=hours_ahead)
        forecast = self.models[upstream].predict(future)
        
        return forecast['yhat'].tail(hours_ahead).values
    
    def detect_anomaly(self, upstream, current_value, prediction):
        """检测当前流量是否异常"""
        threshold = prediction * 1.5  # 超过预测值 50% 视为异常
        return current_value > threshold

from sklearn.preprocessing import LabelEncoder
import tensorflow as tf

class RequestClassifier:
    """请求类型分类器"""
    
    def __init__(self):
        self.model = self._build_model()
        self.label_encoder = LabelEncoder()
    
    def _build_model(self):
        """构建 LSTM 分类模型"""
        model = tf.keras.Sequential([
            tf.keras.layers.LSTM(64, return_sequences=True, input_shape=(10, 8)),
            tf.keras.layers.Dropout(0.2),
            tf.keras.layers.LSTM(32),
            tf.keras.layers.Dense(16, activation='relu'),
            tf.keras.layers.Dense(4, activation='softmax')  # 4 类请求
        ])
        model.compile(optimizer='adam', loss='sparse_categorical_crossentropy')
        return model
    
    def classify(self, request_features):
        """
        分类请求类型
        0: 静态资源 (CSS/JS/Image)
        1: 高频 API (健康检查/状态)
        2: 计算密集型 (查询/分析)
        3: 数据写入 (表单/上传)
        """
        prediction = self.model.predict(request_features.reshape(1, -1, 8))
        return np.argmax(prediction[0])
    
    def get_optimal_backend(self, request_type, backend_metrics):
        """根据请求类型选择最优后端"""
        if request_type == 0:  # 静态资源
            # 选择缓存命中率最高的节点
            return max(backend_metrics, key=lambda x: x['cache_hit_rate'])
        elif request_type == 1:  # 高频 API
            # 选择延迟最低的节点
            return min(backend_metrics, key=lambda x: x['avg_latency'])
        elif request_type == 2:  # 计算密集型
            # 选择 CPU 空闲最多的节点
            return min(backend_metrics, key=lambda x: x['cpu_usage'])
        else:  # 数据写入
            # 分散写入，避免单点压力
            return backend_metrics[0]  # 轮询

-- nginx-lua/upstream_health.lua
local ngx = ngx
local resty = require "resty.core"
local http = require "resty.http"

-- 从外部 API 获取动态权重
function get_dynamic_weights()
    local httpc = http.new()
    local resp, err = httpc:request_uri(
        "http://ai-traffic-controller:8080/api/weights",
        { method = "GET" }
    )
    
    if resp and resp.status == 200 then
        local cjson = require "cjson"
        return cjson.decode(resp.body)
    end
    return nil
end

-- 动态设置上游权重
function set_upstream_weights(backend_list)
    local weights = get_dynamic_weights()
    
    if weights then
        for _, backend in ipairs(backend_list) do
            local weight = weights[backend.server] or 100
            backend.weight = weight
        end
    end
end

-- 健康检查与健康权重
function calculate_healthy_weight(backend)
    local latency = backend.avg_latency
    local error_rate = backend.error_rate
    local active = backend.active_connections
    
    -- 健康分数：低延迟、低错误率、合理负载 = 高权重
    local latency_score = math.max(0, 1 - latency / 5.0)
    local error_score = math.max(0, 1 - error_rate * 10)
    local load_score = math.max(0, 1 - active / 1000)
    
    local health = (latency_score * 0.4 + error_score * 0.4 + load_score * 0.2)
    return math.max(1, math.floor(health * 100))
end

# ai-traffic-controller/main.py
from fastapi import FastAPI
from prometheus_api_client import PrometheusConnect
from datetime import datetime, timedelta
import json

app = FastAPI()
prom = PrometheusConnect(url="http://prometheus:9090")

@app.get("/api/weights")
def get_dynamic_weights():
    """计算并返回动态权重"""
    
    # 1. 获取所有后端状态
    backends = get_backend_status()
    
    # 2. 获取预测数据
    predictions = predict_next_hour()
    
    # 3. AI 决策：计算最优权重
    weights = {}
    for backend in backends:
        base_weight = calculate_health_weight(backend)
        prediction_factor = adjust_for_prediction(backend.name, predictions)
        final_weight = base_weight * prediction_factor
        weights[backend.name] = final_weight
    
    return weights

@app.get("/api/routing-rules")
def get_routing_rules():
    """返回动态路由规则"""
    rules = {
        "static_assets": {
            "pattern": ".*\\.(css|js|png|jpg|svg)$",
            "strategy": "cache-first",
            "backends": get_cdn_backends()
        },
        "api_heavy": {
            "pattern": "/api/v[0-9]+/.*",
            "strategy": "least-loaded",
            "backends": get_api_backends()
        }
    }
    return rules

def calculate_health_weight(backend):
    """基于健康状态的权重计算"""
    latency = backend.get('avg_latency', 0.1)
    error_rate = backend.get('error_rate', 0)
    cpu = backend.get('cpu_usage', 50)
    
    # 延迟贡献：越低越好
    latency_score = max(0, 1 - latency / 2.0)
    # 错误率贡献：越低越好
    error_score = max(0, 1 - error_rate * 20)
    # CPU 贡献：适中最好（太低可能空闲，太高过载）
    cpu_score = 1 - abs(cpu - 60) / 100
    
    return (latency_score * 0.4 + error_score * 0.4 + cpu_score * 0.2)

def adjust_for_prediction(backend_name, predictions):
    """根据预测调整权重"""
    predicted_load = predictions.get(backend_name, 100)
    current_capacity = 1000
    
    # 预测超载时降权
    if predicted_load > current_capacity * 0.8:
        return 0.5
    # 预测空闲时加权
    elif predicted_load < current_capacity * 0.3:
        return 1.5
    return 1.0

from envoy_control.control_plane.control_plane import ControlPlane
from envoy_control.envoy.admin import EnvoyAdmin
from envoy_control.types.cds import CdsUpdate
from envoy_control.types.clusters import Cluster

class EnvoyTrafficManager:
    """使用 Envoy xDS 实现动态负载均衡"""
    
    def __init__(self, envoy_admin_url="http://localhost:9901"):
        self.envoy = EnvoyAdmin(envoy_admin_url)
        self.control_plane = ControlPlane()
    
    def update_cluster_weights(self, cluster_name, weights):
        """动态更新集群权重"""
        new_clusters = []
        for endpoint, weight in weights.items():
            cluster = Cluster(
                name=f"{cluster_name}-{endpoint}",
                lb_type="RING_HASH",  # 一致性哈希
                endpoints=[endpoint],
                health_check=True
            )
            new_clusters.append(cluster)
        
        cds_update = CdsUpdate(new_clusters)
        self.control_plane.push_cds(cds_update)
        self.envoy.post_cds(cds_update.to_json())
    
    def get_endpoints_health(self):
        """获取所有端点健康状态"""
        info = self.envoy.get_info()
        clusters = info.get('clusters', [])
        
        return {
            endpoint['address']: {
                'healthy': endpoint.get('health', 'healthy') == 'healthy',
                'requests_active': endpoint.get('requests_active', 0),
                'requests_completed': endpoint.get('requests_completed', 0)
            }
            for endpoint in clusters
        }

import asyncio
import time
from dataclasses import dataclass
from typing import Dict, List, Optional
from collections import deque

@dataclass
class CircuitBreakerState:
    """熔断器状态"""
    name: str
    state: str  # 'CLOSED', 'OPEN', 'HALF_OPEN'
    failure_count: int
    last_failure_time: float
    half_open_timeout: float = 30.0
    failure_threshold: int = 5
    success_threshold: int = 3

class IntelligentCircuitBreaker:
    """AI 驱动的熔断器"""
    
    def __init__(self, backend_name: str):
        self.name = backend_name
        self.state = CircuitBreakerState(
            name=backend_name,
            state='CLOSED',
            failure_count=0
        )
        self.recent_requests = deque(maxlen=100)
        self.failure_window = deque(maxlen=20)
    
    def record_request(self, success: bool, latency: float):
        """记录请求结果"""
        self.recent_requests.append({
            'success': success,
            'latency': latency,
            'timestamp': time.time()
        })
        
        if not success:
            self.failure_window.append(time.time())
            self.state.failure_count += 1
            self.state.last_failure_time = time.time()
            
            # AI 判断：如果是偶发错误，半开状态测试
            if self._should_half_open():
                self.state.state = 'HALF_OPEN'
        else:
            # 成功请求可能重置计数器
            if self.state.state == 'HALF_OPEN':
                self._handle_half_open_success()
    
    def _should_half_open(self) -> bool:
        """AI 判断是否进入半开状态"""
        # 考虑时间因素：故障间隔是否在增长？
        if len(self.failure_window) >= 5:
            intervals = list(self.failure_window)
            intervals.sort()
            # 如果最近 5 次故障间隔在增加，说明系统可能在恢复
            recent_intervals = [
                intervals[i+1] - intervals[i] 
                for i in range(-5, -1)
            ]
            if all(recent_intervals[i] < recent_intervals[i+1] 
                   for i in range(len(recent_intervals)-1)):
                return True
        return False
    
    def _handle_half_open_success(self):
        """半开状态下成功请求的处理"""
        self.state.failure_count = 0
        self.state.state = 'CLOSED'
        self.failure_window.clear()
    
    def allow_request(self) -> bool:
        """判断是否允许请求通过"""
        if self.state.state == 'CLOSED':
            return True
        elif self.state.state == 'OPEN':
            # 检查熔断时间
            if time.time() - self.state.last_failure_time > self.state.half_open_timeout:
                self.state.state = 'HALF_OPEN'
                return True
            return False
        else:  # HALF_OPEN
            return True  # 允许少量测试请求

class GracefulDegradation:
    """优雅降级策略"""
    
    def __init__(self):
        self.degradation_rules = {
            'high_latency': {
                'condition': 'avg_latency > 500ms',
                'action': 'enable_caching',
                'parameters': {'cache_ttl': 300}
            },
            'high_error_rate': {
                'condition': 'error_rate > 5%',
                'action': 'return_fallback',
                'parameters': {'fallback_data': 'cached_response'}
            },
            'resource_exhaustion': {
                'condition': 'cpu > 90% OR memory > 85%',
                'action': 'reduce_functionality',
                'parameters': {'disable_features': ['real_time_sync', 'batch_export']}
            }
        }
    
    def evaluate_degradation(self, metrics: Dict) -> List[Dict]:
        """评估是否需要降级"""
        actions = []
        
        for rule_name, rule in self.degradation_rules.items():
            if self._evaluate_condition(rule['condition'], metrics):
                actions.append({
                    'rule': rule_name,
                    'action': rule['action'],
                    'parameters': rule['parameters'],
                    'priority': self._calculate_priority(rule_name)
                })
        
        # 按优先级排序
        return sorted(actions, key=lambda x: x['priority'], reverse=True)
    
    def _evaluate_condition(self, condition: str, metrics: Dict) -> bool:
        """评估条件是否满足"""
        # 简化的条件评估
        if '>=' in condition:
            metric_name, threshold = condition.split(' >= ')
            return metrics.get(metric_name, 0) >= float(threshold)
        elif '>' in condition:
            metric_name, threshold = condition.split(' > ')
            return metrics.get(metric_name, 0) > float(threshold)
        return False
    
    def _calculate_priority(self, rule_name: str) -> int:
        """计算降级规则的优先级"""
        priority_map = {
            'resource_exhaustion': 1,  # 最高优先级
            'high_error_rate': 2,
            'high_latency': 3
        }
        return priority_map.get(rule_name, 99)

version: '3.8'

services:
  # AI 控制平面
  ai-controller:
    build: ./ai-controller
    environment:
      - PROMETHEUS_URL=http://prometheus:9090
      - NGINX_API_URL=http://nginx:8080
    volumes:
      - ./ai-controller:/app
    depends_on:
      - prometheus
      - loki
    networks:
      - traffic-net

  # Nginx + Lua
  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "8080:8080"
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/nginx.conf
      - ./nginx/lua:/etc/nginx/lua
    depends_on:
      - ai-controller
    networks:
      - traffic-net

  # 后端服务
  backend-api:
    image: your-api-service:latest
    networks:
      - traffic-net
    deploy:
      replicas: 3

  # 静态资源服务
  backend-cdn:
    image: your-cdn-service:latest
    networks:
      - traffic-net

  # 监控栈
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus/prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus-data:/prometheus
    networks:
      - traffic-net

  loki:
    image: grafana/loki:latest
    command: -config.file=/etc/loki/local-config.yaml
    networks:
      - traffic-net

  vector:
    image: timberio/vector:latest-alpine
    volumes:
      - ./vector/vector.toml:/etc/vector/vector.toml
      - /var/log/nginx:/var/log/nginx:ro
    networks:
      - traffic-net

volumes:
  prometheus-data:

networks:
  traffic-net:
    driver: bridge

# nginx/nginx.conf
worker_processes auto;
error_log /var/log/nginx/error.log warn;
pid /var/run/nginx.pid;

events {
    worker_connections 4096;
}

http {
    # 自定义日志格式
    log_format ai_monitor '$remote_addr - $request_time $upstream_response_time '
                          '$upstream_status $request_length $bytes_sent '
                          '$http_x_forwarded_for $request_method $upstream_addr';
    
    # 上游定义
    upstream api_backend {
        least_conn;
        server backend-api-1:8000 weight=100 max_fails=3;
        server backend-api-2:8000 weight=100 max_fails=3;
        server backend-api-3:8000 weight=100 max_fails=3;
    }
    
    upstream static_backend {
        ip_hash;
        server backend-cdn-1:8000 weight=100;
        server backend-cdn-2:8000 weight=100;
    }
    
    server {
        listen 80;
        server_name _;
        
        # Lua 初始化
        lua_package_path "/etc/nginx/lua/?.lua;;";
        
        # 动态权重获取
        init_by_lua_block {
            ai_controller = require "ai_controller"
        }
        
        # 请求处理
        location / {
            # 获取动态权重
            content_by_lua_block {
                local weights = ai_controller.get_weights()
                if weights then
                    -- 动态调整上游权重
                    for backend, weight in pairs(weights) do
                        -- 设置权重逻辑
                    end
                end
            }
            
            proxy_pass http://api_backend;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
        }
        
        # 静态资源
        location ~* \.(css|js|png|jpg|svg|gif)$ {
            proxy_pass http://static_backend;
            expires 30d;
            add_header Cache-Control "public, immutable";
        }
        
        # AI 控制器 API
        location /api/ai {
            proxy_pass http://ai-controller:8000;
        }
    }
}

# 多层缓存架构
cache_layers = {
    'L1': {  # 本地内存缓存
        'type': 'memory',
        'ttl': 60,
        'max_size': '100MB'
    },
    'L2': {  # Redis 缓存
        'type': 'redis',
        'ttl': 300,
        'max_size': '1GB'
    },
    'L3': {  # 对象存储缓存
        'type': 's3',
        'ttl': 3600,
        'path': '/cached/responses'
    }
}

import asyncio
from aiomysql import create_pool

class IntelligentConnectionPool:
    """智能连接池管理"""
    
    def __init__(self, backend_url, max_connections=100):
        self.backend_url = backend_url
        self.max_connections = max_connections
        self.pool = None
        self.connection_stats = {}
    
    async def get_connection(self, request_type='read'):
        """根据请求类型获取连接"""
        if self.pool is None:
            self.pool = await create_pool(
                db='your_db',
                user='user',
                password='pass',
                host=self.backend_url
            )
        
        # 根据请求类型分配连接
        if request_type == 'write':
            # 写操作需要更强的连接
            return await self._get_strong_connection()
        else:
            # 读操作可以使用缓存连接
            return await self._get_cached_connection()
    
    async def _get_strong_connection(self):
        """获取高性能连接"""
        async with self.pool.get() as conn:
            await conn.execute("SET max_execution_time=5000")
            return conn
    
    async def _get_cached_connection(self):
        """获取缓存友好连接"""
        async with self.pool.get() as conn:
            await conn.execute("SET max_execution_time=1000")
            return conn

# grafana_dashboard.json
{
  "dashboard": {
    "title": "AI Traffic Management Overview",
    "panels": [
      {
        "title": "请求延迟分布",
        "type": "histogram",
        "expr": "histogram_quantile(0.99, sum(rate(nginx_request_duration_seconds_bucket[5m])) by (le, upstream))"
      },
      {
        "title": "错误率趋势",
        "type": "graph",
        "expr": "sum(rate(nginx_http_requests_total{status=~\"5..\"}[5m])) by (upstream)"
      },
      {
        "title": "动态权重变化",
        "type": "graph",
        "expr": "nginx_upstream_weight"
      },
      {
        "title": "流量预测 vs 实际",
        "type": "graph",
        "exprs": [
          "predicted_requests",
          "actual_requests"
        ]
      }
    ]
  }
}

# alert_rules.yml
groups:
  - name: ai_traffic_alerts
    rules:
      - alert: HighLatencyBackends
        expr: ai:request_latency_p99 > 2
        for: 5m
        annotations:
          summary: "后端平均延迟超过 2 秒"
          description: "请检查 AI 控制器权重分配"
          
      - alert: PredictiveCapacityExhaustion
        expr: predicted_load / total_capacity > 0.85
        for: 10m
        annotations:
          summary: "预测容量即将耗尽"
          description: "AI 预测未来 1 小时负载将超 85%"
          
      - alert: CircuitBreakerTripped
        expr: circuit_breaker_state == 1
        for: 1m
        annotations:
          summary: "熔断器触发"
          description: "后端服务 {{ $labels.backend }} 熔断器打开"