142_故障容错：冗余与回滚机制 - 配置多副本的独特健康检查

import numpy as np
import matplotlib.pyplot as plt

# 模拟不同故障类型下的服务可用性
def simulate_service_availability(failure_types, recovery_times):
    availability = []
    for failure, recovery in zip(failure_types, recovery_times):
        # 计算一年中的可用时间比例
        yearly_minutes = 365 * 24 * 60
        downtime = recovery
        avail_percent = (yearly_minutes - downtime) / yearly_minutes * 100
        availability.append(avail_percent)
    
    return availability

# 故障类型和恢复时间（分钟）
failure_types = ['GPU故障', '网络分区', '内存泄漏', '模型错误', '容量不足', '配置错误']
daily_failures = [0.1, 0.02, 0.05, 0.08, 0.2, 0.03]  # 每日故障率
recovery_times = [30, 120, 60, 45, 15, 20]  # 平均恢复时间（分钟）

# 计算年平均停机时间
yearly_downtime = [daily * 365 * recovery for daily, recovery in zip(daily_failures, recovery_times)]
availability = simulate_service_availability(failure_types, yearly_downtime)

# 绘制服务可用性对比图
plt.figure(figsize=(12, 6))
bars = plt.bar(failure_types, availability, color='skyblue')
plt.axhline(y=99.9, color='r', linestyle='--', label='99.9% SLA')
plt.axhline(y=99.99, color='g', linestyle='--', label='99.99% SLA')
plt.ylim(99, 100)
plt.ylabel('服务可用性 (%)')
plt.title('不同故障类型下的LLM服务可用性 (2025年数据)')
plt.xticks(rotation=45)
plt.legend()

# 添加数值标签
for bar, avail in zip(bars, availability):
    height = bar.get_height()
    plt.text(bar.get_x() + bar.get_width()/2., height, 
            f'{avail:.4f}%',
            ha='center', va='bottom')

plt.tight_layout()
plt.show()

import random
import numpy as np
import matplotlib.pyplot as plt

# 模拟不同负载均衡策略的效果
def simulate_load_balancing(servers, requests, strategy):
    server_loads = [0] * servers
    
    for i in range(requests):
        if strategy == 'round_robin':
            # 轮询策略
            target = i % servers
        elif strategy == 'random':
            # 随机策略
            target = random.randint(0, servers-1)
        elif strategy == 'least_connections':
            # 最少连接策略
            target = server_loads.index(min(server_loads))
        elif strategy == 'weighted_round_robin':
            # 加权轮询（模拟GPU性能差异）
            weights = [1.2, 1.0, 0.8, 0.9] if servers >= 4 else [1.0] * servers
            total_weight = sum(weights)
            r = random.uniform(0, total_weight)
            current = 0
            for j in range(servers):
                current += weights[j]
                if r <= current:
                    target = j
                    break
        
        # 模拟请求处理时间（LLM推理时间通常服从长尾分布）
        processing_time = np.random.gamma(shape=2, scale=1.5)
        server_loads[target] += processing_time
    
    return server_loads

# 模拟不同策略下的负载分布
def compare_load_balancing_strategies():
    servers = 4
    requests = 1000
    strategies = ['round_robin', 'random', 'least_connections', 'weighted_round_robin']
    
    plt.figure(figsize=(15, 10))
    
    for i, strategy in enumerate(strategies):
        loads = simulate_load_balancing(servers, requests, strategy)
        
        # 计算负载均衡指标
        mean_load = np.mean(loads)
        std_load = np.std(loads)
        load_imbalance = std_load / mean_load if mean_load > 0 else 0
        
        plt.subplot(2, 2, i+1)
        bars = plt.bar(range(1, servers+1), loads, color='skyblue')
        plt.axhline(y=mean_load, color='r', linestyle='--', label=f'Mean: {mean_load:.2f}')
        plt.title(f'{strategy.replace("_", " ").title()}\nImbalance Index: {load_imbalance:.4f}')
        plt.xlabel('Server')
        plt.ylabel('Total Load')
        plt.xticks(range(1, servers+1))
        plt.legend()
        
        # 添加数值标签
        for bar, load in zip(bars, loads):
            height = bar.get_height()
            plt.text(bar.get_x() + bar.get_width()/2., height, 
                    f'{load:.2f}',
                    ha='center', va='bottom')
    
    plt.tight_layout()
    plt.show()
    
    print("负载均衡策略对比完成")

# 运行模拟
compare_load_balancing_strategies()```

从模拟结果可以看出，最少连接策略和加权轮询策略在处理LLM请求时通常能提供更好的负载均衡效果，特别是当服务器之间存在性能差异时。

### 2.3 Kubernetes中的多副本部署

Kubernetes作为容器编排平台，提供了强大的多副本管理能力。下面我们来看看如何在Kubernetes中部署具有高可用性的LLM服务：

#### 部署清单示例

```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-inference-service
spec:
  replicas: 3  # 3个副本实例
  selector:
    matchLabels:
      app: llm-service
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxUnavailable: 1  # 更新过程中最多允许1个实例不可用
      maxSurge: 1        # 更新过程中最多允许增加1个实例
  template:
    metadata:
      labels:
        app: llm-service
    spec:
      affinity:
        podAntiAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 100
            podAffinityTerm:
              labelSelector:
                matchExpressions:
                - key: app
                  operator: In
                  values:
                  - llm-service
              topologyKey: "kubernetes.io/hostname"  # 尽量将pod分散到不同节点
      containers:
      - name: llm-inference
        image: llm-inference:latest
        resources:
          requests:
            memory: "16Gi"
            cpu: "4"
            nvidia.com/gpu: 1
          limits:
            memory: "24Gi"
            cpu: "8"
            nvidia.com/gpu: 1
        ports:
        - containerPort: 8000
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        livenessProbe:
          httpGet:
            path: /live
            port: 8000
          initialDelaySeconds: 60
          periodSeconds: 20

from fastapi import FastAPI, HTTPException, Response
from fastapi.middleware.cors import CORSMiddleware
import time
import torch
import psutil
import threading
import asyncio
from transformers import AutoModelForCausalLM, AutoTokenizer
import gc
import os
import prometheus_client
from prometheus_client import Counter, Histogram, Gauge, start_http_server

app = FastAPI()

# 配置CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境中应该设置具体的允许来源
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 模拟模型加载
try:
    # 在实际部署中，这里会加载真实的LLM模型
    # model = AutoModelForCausalLM.from_pretrained("your-model-name")
    # tokenizer = AutoTokenizer.from_pretrained("your-model-name")
    MODEL_LOADED = True
    print("Model simulated as loaded")
except Exception as e:
    MODEL_LOADED = False
    print(f"Failed to load model: {str(e)}")

# 模拟依赖服务状态
DEPENDENCIES_HEALTHY = {
    "database": True,
    "vector_store": True,
    "cache_service": True
}

# 初始化Prometheus指标
REQUEST_COUNT = Counter('llm_requests_total', 'Total number of LLM requests', ['endpoint', 'status'])
INFERENCE_TIME = Histogram('llm_inference_seconds', 'Time spent processing inference requests')
GPU_MEMORY_USAGE = Gauge('llm_gpu_memory_bytes', 'GPU memory usage in bytes')
CPU_USAGE = Gauge('llm_cpu_usage_percent', 'CPU usage percentage')
MEMORY_USAGE = Gauge('llm_memory_usage_bytes', 'Memory usage in bytes')
MODEL_LOAD_STATUS = Gauge('llm_model_loaded', 'Model load status (1=loaded, 0=not loaded)')

# 设置初始模型加载状态
MODEL_LOAD_STATUS.set(1 if MODEL_LOADED else 0)

# 后台线程用于更新系统指标
def update_system_metrics():
    """定期更新系统资源使用指标"""
    while True:
        # 更新CPU使用率
        CPU_USAGE.set(psutil.cpu_percent(interval=1))
        
        # 更新内存使用率
        memory_info = psutil.virtual_memory()
        MEMORY_USAGE.set(memory_info.used)
        
        # 尝试更新GPU内存使用情况
        try:
            if torch.cuda.is_available():
                GPU_MEMORY_USAGE.set(torch.cuda.max_memory_allocated())
        except Exception as e:
            print(f"Error updating GPU metrics: {str(e)}")
        
        time.sleep(10)  # 每10秒更新一次

# 启动指标更新线程
metrics_thread = threading.Thread(target=update_system_metrics, daemon=True)
metrics_thread.start()

# 启动Prometheus指标导出器
prometheus_port = int(os.environ.get('PROMETHEUS_PORT', 8001))
prometheus_thread = threading.Thread(
    target=start_http_server, 
    args=(prometheus_port,),
    daemon=True
)
prometheus_thread.start()

print(f"Prometheus metrics available at http://localhost:{prometheus_port}")

# 健康检查端点
@app.get("/health/liveness")
async def liveness_check():
    """基本的存活检查，验证服务是否在运行"""
    return {"status": "healthy", "timestamp": time.time()}

@app.get("/health/readiness")
async def readiness_check():
    """就绪检查，验证服务是否可以接受请求"""
    # 检查模型是否加载
    if not MODEL_LOADED:
        REQUEST_COUNT.labels(endpoint="readiness", status="unhealthy").inc()
        return Response(
            content={"status": "unhealthy", "reason": "model_not_loaded"}, 
            status_code=503, 
            media_type="application/json"
        )
    
    # 检查所有依赖服务
    for service, healthy in DEPENDENCIES_HEALTHY.items():
        if not healthy:
            REQUEST_COUNT.labels(endpoint="readiness", status="unhealthy").inc()
            return Response(
                content={"status": "unhealthy", "reason": f"dependency_{service}_failed"}, 
                status_code=503,
                media_type="application/json"
            )
    
    # 检查系统资源是否充足
    memory_info = psutil.virtual_memory()
    if memory_info.percent > 90:
        REQUEST_COUNT.labels(endpoint="readiness", status="unhealthy").inc()
        return Response(
            content={"status": "unhealthy", "reason": "high_memory_usage"}, 
            status_code=503,
            media_type="application/json"
        )
    
    if psutil.cpu_percent(interval=1) > 90:
        REQUEST_COUNT.labels(endpoint="readiness", status="unhealthy").inc()
        return Response(
            content={"status": "unhealthy", "reason": "high_cpu_usage"}, 
            status_code=503,
            media_type="application/json"
        )
    
    REQUEST_COUNT.labels(endpoint="readiness", status="healthy").inc()
    return {"status": "healthy", "timestamp": time.time()}

@app.get("/health/startup")
async def startup_check():
    """启动检查，验证服务是否已完全启动"""
    # 检查模型是否加载完成
    if not MODEL_LOADED:
        return Response(
            content={"status": "unhealthy", "reason": "model_not_loaded"}, 
            status_code=503,
            media_type="application/json"
        )
    
    # 可以添加其他启动验证逻辑
    return {"status": "healthy", "timestamp": time.time(), "version": "1.0.0"}

@app.get("/health/deep")
async def deep_health_check():
    """深度健康检查，执行模型推理测试"""
    start_time = time.time()
    
    try:
        # 执行一个简单的模型推理测试
        # 在实际部署中，这里会使用真实模型进行推理
        # 模拟推理过程
        await asyncio.sleep(0.5)
        
        inference_time = time.time() - start_time
        INFERENCE_TIME.observe(inference_time)
        
        # 验证推理结果
        test_passed = True  # 模拟测试通过
        
        if test_passed:
            REQUEST_COUNT.labels(endpoint="deep", status="healthy").inc()
            return {
                "status": "healthy",
                "timestamp": time.time(),
                "inference_time_ms": inference_time * 1000,
                "model_status": "operational"
            }
        else:
            REQUEST_COUNT.labels(endpoint="deep", status="unhealthy").inc()
            return Response(
                content={"status": "unhealthy", "reason": "inference_test_failed"},
                status_code=503,
                media_type="application/json"
            )
    except Exception as e:
        REQUEST_COUNT.labels(endpoint="deep", status="error").inc()
        return Response(
            content={"status": "unhealthy", "reason": f"deep_check_error: {str(e)}"},
            status_code=503,
            media_type="application/json"
        )

@app.post("/health/trigger_gc")
async def trigger_garbage_collection():
    """手动触发垃圾回收，用于清理资源"""
    try:
        # 执行Python垃圾回收
        gc.collect()
        
        # 如果有GPU，尝试清理GPU内存
        if torch.cuda.is_available():
            torch.cuda.empty_cache()
            torch.cuda.ipc_collect()
        
        return {"status": "success", "message": "Garbage collection triggered"}
    except Exception as e:
        return Response(
            content={"status": "error", "message": f"Failed to trigger GC: {str(e)}"},
            status_code=500,
            media_type="application/json"
        )

@app.get("/health/metrics")
async def get_health_metrics():
    """获取服务的健康指标摘要"""
    try:
        memory_info = psutil.virtual_memory()
        cpu_percent = psutil.cpu_percent(interval=1)
        
        metrics = {
            "system": {
                "cpu_percent": cpu_percent,
                "memory_percent": memory_info.percent,
                "memory_used_mb": round(memory_info.used / (1024 * 1024), 2),
                "memory_total_mb": round(memory_info.total / (1024 * 1024), 2)
            },
            "model": {
                "loaded": MODEL_LOADED,
                "version": "1.0.0"  # 实际部署中应从配置或环境变量获取
            },
            "dependencies": DEPENDENCIES_HEALTHY.copy(),
            "timestamp": time.time()
        }
        
        # 添加GPU信息（如果可用）
        if torch.cuda.is_available():
            gpu_count = torch.cuda.device_count()
            gpu_metrics = []
            
            for i in range(gpu_count):
                gpu_properties = torch.cuda.get_device_properties(i)
                gpu_metrics.append({
                    "name": gpu_properties.name,
                    "memory_allocated_mb": round(torch.cuda.memory_allocated(i) / (1024 * 1024), 2),
                    "memory_reserved_mb": round(torch.cuda.memory_reserved(i) / (1024 * 1024), 2),
                    "total_memory_mb": round(gpu_properties.total_memory / (1024 * 1024), 2)
                })
            
            metrics["gpu"] = gpu_metrics
        
        return metrics
    except Exception as e:
        return Response(
            content={"status": "error", "message": f"Failed to collect metrics: {str(e)}"},
            status_code=500,
            media_type="application/json"
        )

# 模拟LLM推理端点
@app.post("/api/inference")
async def llm_inference(prompt: str, max_tokens: int = 100):
    """LLM推理端点，使用健康检查中监控的指标"""
    try:
        start_time = time.time()
        
        # 检查模型状态
        if not MODEL_LOADED:
            REQUEST_COUNT.labels(endpoint="inference", status="error").inc()
            raise HTTPException(status_code=503, detail="Model not loaded")
        
        # 执行推理（模拟）
        await asyncio.sleep(1.0)  # 模拟推理延迟
        
        inference_time = time.time() - start_time
        INFERENCE_TIME.observe(inference_time)
        REQUEST_COUNT.labels(endpoint="inference", status="success").inc()
        
        # 模拟推理结果
        return {
            "prompt": prompt,
            "generated_text": f"Simulated response for: {prompt}",
            "inference_time_ms": inference_time * 1000,
            "tokens_generated": min(max_tokens, 50)  # 模拟生成的tokens数
        }
    except HTTPException:
        raise
    except Exception as e:
        REQUEST_COUNT.labels(endpoint="inference", status="error").inc()
        raise HTTPException(status_code=500, detail=f"Inference error: {str(e)}")

# 依赖服务健康管理端点
@app.post("/admin/dependency/{service_name}")
async def set_dependency_health(service_name: str, healthy: bool):
    """管理依赖服务健康状态（用于测试）"""
    if service_name in DEPENDENCIES_HEALTHY:
        DEPENDENCIES_HEALTHY[service_name] = healthy
        return {"status": "success", f"{service_name}_healthy": healthy}
    else:
        raise HTTPException(status_code=404, detail=f"Dependency {service_name} not found")

# 优雅关闭处理
@app.on_event("shutdown")
async def shutdown_event():
    """优雅关闭服务时执行的清理操作"""
    print("Shutting down LLM service...")
    
    # 清理资源
    if torch.cuda.is_available():
        torch.cuda.empty_cache()
        torch.cuda.ipc_collect()
    
    gc.collect()
    print("Resources cleaned up")

# 主程序入口
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
# 配置CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 全局变量
model_health = {
    "status": "initializing",
    "last_inference_time": 0,
    "error_count": 0,
    "memory_usage": 0,
    "gpu_utilization": 0,
}

# 加载模型（在实际应用中，这通常在启动时完成）
def load_model():
    try:
        # 模拟模型加载
        print("Loading model...")
        time.sleep(5)  # 模拟加载时间
        # model = AutoModelForCausalLM.from_pretrained("your-model")
        # tokenizer = AutoTokenizer.from_pretrained("your-model")
        model_health["status"] = "healthy"
        print("Model loaded successfully")
    except Exception as e:
        model_health["status"] = "unhealthy"
        model_health["error_count"] += 1
        print(f"Error loading model: {e}")

# 启动时加载模型
threading.Thread(target=load_model).start()

# 模拟推理函数
def inference(text):
    try:
        start_time = time.time()
        # 在实际应用中，这里会调用模型进行推理
        time.sleep(0.5)  # 模拟推理时间
        end_time = time.time()
        
        # 更新健康指标
        model_health["last_inference_time"] = end_time - start_time
        model_health["error_count"] = 0  # 重置错误计数
        
        # 模拟GPU利用率
        model_health["gpu_utilization"] = min(100, model_health["gpu_utilization"] + 10)
        
        return {"text": text, "generated": "This is a simulated response"}
    except Exception as e:
        model_health["error_count"] += 1
        if model_health["error_count"] > 3:  # 连续错误超过阈值，标记为不健康
            model_health["status"] = "unhealthy"
        raise HTTPException(status_code=500, detail=str(e))

# 基础健康检查（就绪探针）
@app.get("/health")
def health_check():
    # 检查进程状态
    process = psutil.Process()
    memory_info = process.memory_info()
    model_health["memory_usage"] = memory_info.rss / (1024 * 1024 * 1024)  # 转换为GB
    
    # 检查内存使用
    if model_health["memory_usage"] > 20:  # 假设内存限制为20GB
        model_health["status"] = "degraded"
    
    # 返回详细的健康状态
    return {
        "status": model_health["status"],
        "memory_usage_gb": round(model_health["memory_usage"], 2),
        "error_count": model_health["error_count"],
        "last_inference_time_ms": round(model_health["last_inference_time"] * 1000, 2),
        "timestamp": time.time()
    }

# 存活探针（更基础的检查）
@app.get("/live")
def live_check():
    if model_health["status"] == "unhealthy":
        return Response(status_code=503, content="Service Unhealthy")
    return {"status": "alive"}

# 深度健康检查（业务逻辑检查）
@app.get("/deep_health")
async def deep_health_check():
    try:
        # 执行一个简单的推理测试
        result = inference("Test health check")
        
        # 检查依赖服务（如果有）
        # 这里可以添加数据库连接检查、缓存服务检查等
        
        return {
            "status": "healthy",
            "inference_test": "passed",
            "dependencies": "all_available",
            "timestamp": time.time()
        }
    except Exception as e:
        return Response(status_code=503, content=f"Deep health check failed: {str(e)}")

# 业务接口
@app.post("/inference")
async def run_inference(request: dict):
    text = request.get("text", "")
    if not text:
        raise HTTPException(status_code=400, detail="Text is required")
    
    return inference(text)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

import time
import random
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta

# 模拟健康数据收集
def collect_health_metrics(instances=5, duration_hours=24):
    metrics = []
    status_options = ["healthy", "degraded", "unhealthy"]
    
    # 生成过去24小时的数据
    now = datetime.now()
    timestamps = [now - timedelta(minutes=i) for i in range(duration_hours*60)]
    timestamps.reverse()  # 从早到晚
    
    for instance_id in range(1, instances + 1):
        # 每个实例的基准状态是健康的，但会随机出现一些问题
        base_status = "healthy"
        
        for timestamp in timestamps:
            # 模拟状态变化（较低概率）
            if random.random() < 0.001:  # 0.1%的概率状态会变化
                base_status = random.choice(status_options)
            
            # 根据状态生成相应的指标
            if base_status == "healthy":
                memory_usage = random.uniform(10, 15)  # GB
                inference_time = random.uniform(0.2, 0.8)  # seconds
                error_count = 0
            elif base_status == "degraded":
                memory_usage = random.uniform(16, 19)  # GB
                inference_time = random.uniform(0.9, 1.5)  # seconds
                error_count = random.randint(1, 3)
            else:  # unhealthy
                memory_usage = random.uniform(19, 22)  # GB
                inference_time = random.uniform(1.6, 3.0)  # seconds
                error_count = random.randint(4, 10)
            
            metrics.append({
                "timestamp": timestamp,
                "instance_id": f"instance-{instance_id}",
                "status": base_status,
                "memory_usage_gb": round(memory_usage, 2),
                "inference_time_ms": round(inference_time * 1000, 2),
                "error_count": error_count
            })
    
    return pd.DataFrame(metrics)

# 生成健康数据
df = collect_health_metrics()

# 1. 实例状态分布图
plt.figure(figsize=(12, 6))
status_counts = df.groupby(["instance_id", "status"]).size().unstack(fill_value=0)
status_counts.plot(kind="bar", stacked=True, ax=plt.gca())
plt.title("LLM服务实例状态分布")
plt.xlabel("实例ID")
plt.ylabel("状态持续时间（分钟）")
plt.legend(title="状态")
plt.tight_layout()
plt.show()

# 2. 内存使用趋势图
plt.figure(figsize=(14, 7))
for instance_id in df["instance_id"].unique():
    instance_data = df[df["instance_id"] == instance_id]
    # 每小时取一个数据点，避免过于密集
    sampled_data = instance_data.iloc[::60]
    plt.plot(sampled_data["timestamp"], sampled_data["memory_usage_gb"], label=instance_id)

plt.axhline(y=18, color='r', linestyle='--', label='警告阈值')
plt.axhline(y=20, color='darkred', linestyle='--', label='临界阈值')
plt.title("LLM服务内存使用趋势")
plt.xlabel("时间")
plt.ylabel("内存使用 (GB)")
plt.legend()
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()

# 3. 推理时间热图
# 按小时和实例ID聚合
pivot_data = df.copy()
pivot_data["hour"] = pivot_data["timestamp"].dt.hour
pivot_data["date"] = pivot_data["timestamp"].dt.date

# 取最近一天的数据
latest_date = pivot_data["date"].max()
daily_data = pivot_data[pivot_data["date"] == latest_date]

# 计算每小时每个实例的平均推理时间
hourly_avg = daily_data.groupby(["hour", "instance_id"])['inference_time_ms'].mean().reset_index()
pivot_table = hourly_avg.pivot(index="hour", columns="instance_id", values="inference_time_ms")

plt.figure(figsize=(12, 8))
sns.heatmap(pivot_table, annot=True, cmap="YlOrRd", fmt=".1f", cbar_kws={'label': '推理时间 (ms)'})
plt.title(f"{latest_date} LLM服务推理时间热图")
plt.xlabel("实例ID")
plt.ylabel("小时")
plt.tight_layout()
plt.show()

import numpy as np
import matplotlib.pyplot as plt
from scipy import stats

# 生成正常数据（模拟LLM服务的响应时间）
def generate_normal_data(n=1000, base=200, noise=20):
    return base + np.random.normal(0, noise, n)

# 生成异常数据
def inject_anomalies(data, anomaly_ratio=0.05):
    data_copy = data.copy()
    num_anomalies = int(len(data) * anomaly_ratio)
    
    # 随机选择异常点位置
    anomaly_indices = np.random.choice(len(data), num_anomalies, replace=False)
    
    # 注入两种类型的异常：突增和持续增长
   突增_indices = anomaly_indices[:len(anomaly_indices)//2]
    持续增长_indices = anomaly_indices[len(anomaly_indices)//2:]
    
    # 突增异常
    data_copy[突增_indices] *= np.random.uniform(2, 5, len(突增_indices))
    
    # 持续增长异常（创建一个趋势）
    for i in 持续增长_indices:
        # 从i开始的一段数据持续增长
        end = min(i + np.random.randint(5, 20), len(data))
        data_copy[i:end] = data_copy[i:end] * (1 + np.linspace(0, 3, end-i))
    
    return data_copy, anomaly_indices

# Z-Score异常检测
def zscore_detection(data, threshold=3):
    z_scores = np.abs(stats.zscore(data))
    anomalies = np.where(z_scores > threshold)[0]
    return anomalies

# 移动平均异常检测
def moving_average_detection(data, window_size=20, threshold=2):
    moving_avg = np.convolve(data, np.ones(window_size)/window_size, mode='same')
    moving_std = np.array([np.std(data[max(0, i-window_size//2):min(len(data), i+window_size//2)]) 
                          for i in range(len(data))])
    
    # 计算偏差倍数
    deviations = np.abs(data - moving_avg) / (moving_std + 1e-10)  # 添加小值避免除零
    anomalies = np.where(deviations > threshold)[0]
    return anomalies

# CUSUM算法
def cusum_detection(data, threshold=50, drift=1):
    n = len(data)
    upper_cusum = np.zeros(n)
    lower_cusum = np.zeros(n)
    
    for i in range(1, n):
        upper_cusum[i] = max(0, upper_cusum[i-1] + data[i] - np.mean(data[:i]) - drift)
        lower_cusum[i] = max(0, lower_cusum[i-1] + np.mean(data[:i]) - data[i] - drift)
    
    anomalies = np.where((upper_cusum > threshold) | (lower_cusum > threshold))[0]
    return anomalies

# 运行实验
normal_data = generate_normal_data()
anomalous_data, actual_anomalies = inject_anomalies(normal_data)

# 检测异常
zscore_anomalies = zscore_detection(anomalous_data)
ma_anomalies = moving_average_detection(anomalous_data)
cusum_anomalies = cusum_detection(anomalous_data)

# 评估检测效果
def evaluate_detection(predicted, actual):
    # 计算真阳性（正确检测到的异常）
    true_positives = len(set(predicted) & set(actual))
    # 计算假阳性（误报）
    false_positives = len(set(predicted) - set(actual))
    # 计算假阴性（漏报）
    false_negatives = len(set(actual) - set(predicted))
    
    precision = true_positives / (true_positives + false_positives + 1e-10)
    recall = true_positives / (true_positives + false_negatives + 1e-10)
    f1_score = 2 * (precision * recall) / (precision + recall + 1e-10)
    
    return {
        "precision": precision,
        "recall": recall,
        "f1_score": f1_score
    }

zscore_metrics = evaluate_detection(zscore_anomalies, actual_anomalies)
ma_metrics = evaluate_detection(ma_anomalies, actual_anomalies)
cusum_metrics = evaluate_detection(cusum_anomalies, actual_anomalies)

# 打印评估结果
print("Z-Score 检测效果:", zscore_metrics)
print("移动平均 检测效果:", ma_metrics)
print("CUSUM 检测效果:", cusum_metrics)

# 可视化结果
plt.figure(figsize=(15, 10))

# 绘制数据和异常点
plt.subplot(2, 1, 1)
plt.plot(anomalous_data, label="Response Time")
plt.scatter(actual_anomalies, anomalous_data[actual_anomalies], color='red', label="Actual Anomalies")
plt.title("LLM Service Response Time with Anomalies")
plt.xlabel("Sample Index")
plt.ylabel("Response Time (ms)")
plt.legend()

# 绘制不同算法的检测结果
plt.subplot(2, 1, 2)
plt.plot(anomalous_data, label="Response Time")
plt.scatter(zscore_anomalies, anomalous_data[zscore_anomalies], color='green', label="Z-Score Detection")
plt.scatter(ma_anomalies, anomalous_data[ma_anomalies], color='orange', label="Moving Average Detection")
plt.scatter(cusum_anomalies, anomalous_data[cusum_anomalies], color='purple', label="CUSUM Detection")
plt.title("Anomaly Detection Results Comparison")
plt.xlabel("Sample Index")
plt.ylabel("Response Time (ms)")
plt.legend()

plt.tight_layout()
plt.show()

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: llm-inference-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: llm-inference-service
  minReplicas: 3  # 最小副本数
  maxReplicas: 10  # 最大副本数
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80
  - type: Pods
    pods:
      metric:
        name: inference_requests_per_second
      target:
        type: AverageValue
        averageValue: 10  # 每个Pod每秒处理10个请求
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 60  # 扩容稳定窗口
      policies:
      - type: Percent
        value: 100
        periodSeconds: 60
    scaleDown:
      stabilizationWindowSeconds: 300  # 缩容稳定窗口（更长，避免频繁缩容）
      policies:
      - type: Percent
        value: 20
        periodSeconds: 300

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from statsmodels.tsa.arima.model import ARIMA
from sklearn.metrics import mean_absolute_error

# 生成模拟的LLM服务请求数据（包含周期性模式）
def generate_request_data(days=7, hourly_pattern=True, weekly_pattern=True):
    # 生成24*7小时的数据
    hours = days * 24
    timestamps = pd.date_range(start='2025-01-01', periods=hours, freq='H')
    
    # 基础请求数
    base_requests = 100 + np.random.normal(0, 20, hours)
    
    # 添加小时模式（一天内的变化）
    if hourly_pattern:
        hourly_factor = []
        for i in range(hours):
            hour = i % 24
            # 模拟一天的模式：早上和晚上是高峰期
            if 8 <= hour <= 10 or 18 <= hour <= 22:
                hourly_factor.append(1.5 + np.random.normal(0, 0.1))
            # 凌晨是低谷期
            elif 0 <= hour <= 6:
                hourly_factor.append(0.5 + np.random.normal(0, 0.1))
            else:
                hourly_factor.append(1.0 + np.random.normal(0, 0.1))
        hourly_factor = np.array(hourly_factor)
        base_requests *= hourly_factor
    
    # 添加周模式
    if weekly_pattern:
        weekly_factor = []
        for i in range(hours):
            day = (i // 24) % 7
            # 周末请求量更高
            if day == 5 or day == 6:  # 周六、周日
                weekly_factor.append(1.3 + np.random.normal(0, 0.1))
            else:
                weekly_factor.append(1.0 + np.random.normal(0, 0.1))
        weekly_factor = np.array(weekly_factor)
        base_requests *= weekly_factor
    
    # 确保请求数为正
    base_requests = np.maximum(10, base_requests)
    
    # 创建DataFrame
    df = pd.DataFrame({
        'timestamp': timestamps,
        'requests': base_requests
    })
    
    return df

# 预测未来24小时的请求量
def predict_requests(data, forecast_hours=24):
    # 使用最后一周的数据进行训练
    train_data = data['requests'].values[-24*7:]
    
    # 拟合ARIMA模型
    model = ARIMA(train_data, order=(5,1,0))  # ARIMA参数可能需要调优
    model_fit = model.fit()
    
    # 预测
    forecast = model_fit.forecast(steps=forecast_hours)
    
    # 生成预测时间戳
    last_timestamp = data['timestamp'].iloc[-1]
    forecast_timestamps = [last_timestamp + pd.Timedelta(hours=i+1) for i in range(forecast_hours)]
    
    return pd.DataFrame({
        'timestamp': forecast_timestamps,
        'predicted_requests': forecast
    })

# 根据预测请求量计算需要的实例数
def calculate_required_instances(predicted_requests, requests_per_instance_per_hour=1000):
    # 假设每个实例每小时可以处理1000个请求
    instances = np.ceil(predicted_requests / requests_per_instance_per_hour)
    # 确保最小实例数
    return np.maximum(3, instances)  # 最小3个实例

# 生成模拟数据
data = generate_request_data(days=14)  # 生成14天的数据

# 预测未来24小时
forecast = predict_requests(data)

# 计算所需实例数
forecast['required_instances'] = calculate_required_instances(forecast['predicted_requests'])

# 可视化结果
plt.figure(figsize=(15, 10))

# 绘制历史数据和预测数据
plt.subplot(2, 1, 1)
plt.plot(data['timestamp'], data['requests'], label='Historical Requests')
plt.plot(forecast['timestamp'], forecast['predicted_requests'], 'r--', label='Predicted Requests')
plt.title('LLM Service Request Forecast')
plt.xlabel('Timestamp')
plt.ylabel('Requests per Hour')
plt.legend()

# 绘制所需实例数
plt.subplot(2, 1, 2)
plt.step(forecast['timestamp'], forecast['required_instances'], where='post')
plt.title('Predicted Required Instances')
plt.xlabel('Timestamp')
plt.ylabel('Number of Instances')
plt.ylim(bottom=0)

plt.tight_layout()
plt.show()

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: isolate-faulty-pod
spec:
  podSelector:
    matchLabels:
      app: llm-service
      health-status: degraded
  policyTypes:
  - Ingress
  ingress:
  - from:
    - podSelector:
        matchLabels:
          app: health-monitor

import os
import json
import shutil
from datetime import datetime

class ModelVersionManager:
    def __init__(self, base_dir="./model_versions"):
        self.base_dir = base_dir
        self.versions_dir = os.path.join(base_dir, "versions")
        self.metadata_file = os.path.join(base_dir, "metadata.json")
        
        # 初始化目录
        os.makedirs(self.versions_dir, exist_ok=True)
        
        # 初始化元数据
        if os.path.exists(self.metadata_file):
            with open(self.metadata_file, 'r') as f:
                self.metadata = json.load(f)
        else:
            self.metadata = {
                "versions": {},
                "current_version": None,
                "previous_version": None,
                "production_versions": []
            }
            self._save_metadata()
    
    def _save_metadata(self):
        with open(self.metadata_file, 'w') as f:
            json.dump(self.metadata, f, indent=2)
    
    def register_model(self, model_path, version_name=None, description=""):
        # 如果没有指定版本名，使用时间戳
        if version_name is None:
            version_name = f"v{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        
        # 创建版本目录
        version_dir = os.path.join(self.versions_dir, version_name)
        os.makedirs(version_dir, exist_ok=True)
        
        # 复制模型文件
        if os.path.isdir(model_path):
            for item in os.listdir(model_path):
                s = os.path.join(model_path, item)
                d = os.path.join(version_dir, item)
                if os.path.isdir(s):
                    shutil.copytree(s, d, dirs_exist_ok=True)
                else:
                    shutil.copy2(s, d)
        else:
            shutil.copy2(model_path, version_dir)
        
        # 更新元数据
        self.metadata["versions"][version_name] = {
            "created_at": datetime.now().isoformat(),
            "description": description,
            "path": version_dir
        }
        
        self._save_metadata()
        print(f"Model version {version_name} registered successfully")
        return version_name
    
    def list_versions(self):
        versions = list(self.metadata["versions"].items())
        # 按创建时间排序
        versions.sort(key=lambda x: x[1]["created_at"], reverse=True)
        
        print("Available model versions:")
        print(f"{'Version':<20} {'Created At':<25} {'Description'}")
        print("-" * 80)
        
        for version, info in versions:
            current_marker = "[CURRENT]" if version == self.metadata["current_version"] else ""
            production_marker = "[PRODUCTION]" if version in self.metadata["production_versions"] else ""
            print(f"{version:<20} {info['created_at']:<25} {info['description']} {current_marker} {production_marker}")
        
        return versions
    
    def set_current_version(self, version_name):
        if version_name not in self.metadata["versions"]:
            raise ValueError(f"Version {version_name} does not exist")
        
        # 保存当前版本为之前的版本
        self.metadata["previous_version"] = self.metadata["current_version"]
        self.metadata["current_version"] = version_name
        
        self._save_metadata()
        print(f"Current model version set to {version_name}")
    
    def mark_as_production(self, version_name):
        if version_name not in self.metadata["versions"]:
            raise ValueError(f"Version {version_name} does not exist")
        
        if version_name not in self.metadata["production_versions"]:
            self.metadata["production_versions"].append(version_name)
            self._save_metadata()
            print(f"Version {version_name} marked as production")
    
    def rollback(self, version_name=None):
        # 如果没有指定版本，回滚到上一个版本
        if version_name is None:
            if self.metadata["previous_version"] is None:
                raise ValueError("No previous version to rollback to")
            version_name = self.metadata["previous_version"]
        
        if version_name not in self.metadata["versions"]:
            raise ValueError(f"Version {version_name} does not exist")
        
        # 保存当前版本
        old_current = self.metadata["current_version"]
        
        # 执行回滚
        self.metadata["previous_version"] = old_current
        self.metadata["current_version"] = version_name
        
        self._save_metadata()
        print(f"Rolled back from {old_current} to {version_name}")
        return version_name
    
    def get_current_model_path(self):
        if self.metadata["current_version"] is None:
            raise ValueError("No current version set")
        
        return self.metadata["versions"][self.metadata["current_version"]]["path"]

# 使用示例
if __name__ == "__main__":
    # 初始化版本管理器
    manager = ModelVersionManager()
    
    # 注册新版本（在实际应用中，这里应该指向真实的模型文件或目录）
    # manager.register_model("./my_latest_model", description="Base LLM model")
    # manager.register_model("./my_improved_model", description="Improved LLM with better performance")
    
    # 列出所有版本
    # manager.list_versions()
    
    # 回滚操作示例
    # manager.rollback()

### 5.2 Kubernetes中的滚动更新与回滚

Kubernetes提供了强大的滚动更新和回滚功能，可以在不中断服务的情况下更新应用，并在出现问题时快速回滚：

#### 执行滚动更新

```bash
# 更新Deployment的镜像版本
kubectl set image deployment/llm-inference-service llm-inference=llm-inference:v2.0

# 或者使用edit命令修改其他配置
kubectl edit deployment/llm-inference-service

# 查看Deployment状态
kubectl rollout status deployment/llm-inference-service

# 查看滚动更新历史
kubectl rollout history deployment/llm-inference-service

# 回滚到上一个版本
kubectl rollout undo deployment/llm-inference-service

# 回滚到指定版本
kubectl rollout undo deployment/llm-inference-service --to-revision=2

# 金丝雀发布配置示例
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: llm-canary-ingress
  annotations:
    nginx.ingress.kubernetes.io/canary: "true"
    nginx.ingress.kubernetes.io/canary-weight: "10"  # 10%的流量路由到金丝雀版本
spec:
  rules:
  - host: llm.example.com
    http:
      paths:
      - path: /
        pathType: Prefix
        backend:
          service:
            name: llm-inference-service-canary
            port:
              number: 8000

import subprocess
import time

def blue_green_deployment():
    # 假设当前运行的是蓝色环境（blue）
    print("Starting blue-green deployment...")
    
    # 1. 部署新版本到绿色环境
    print("Deploying new version to green environment...")
    subprocess.run([
        "kubectl", "apply", "-f", "green-deployment.yaml"
    ], check=True)
    
    # 2. 等待绿色环境就绪
    print("Waiting for green environment to be ready...")
    while True:
        result = subprocess.run(
            ["kubectl", "get", "deployment", "llm-inference-green", "-o", "jsonpath={.status.readyReplicas}"],
            capture_output=True, text=True
        )
        ready_replicas = int(result.stdout) if result.stdout else 0
        
        result = subprocess.run(
            ["kubectl", "get", "deployment", "llm-inference-green", "-o", "jsonpath={.spec.replicas}"],
            capture_output=True, text=True
        )
        total_replicas = int(result.stdout)
        
        if ready_replicas >= total_replicas:
            print("Green environment is ready!")
            break
        
        print(f"Progress: {ready_replicas}/{total_replicas} replicas ready...")
        time.sleep(5)
    
    # 3. 执行健康检查
    print("Performing health checks on green environment...")
    # 这里应该添加实际的健康检查逻辑
    time.sleep(10)  # 模拟健康检查
    
    # 4. 切换流量到绿色环境
    print("Switching traffic to green environment...")
    subprocess.run([
        "kubectl", "apply", "-f", "green-service.yaml"
    ], check=True)
    
    # 5. 监控一段时间
    print("Monitoring green environment...")
    time.sleep(30)  # 监控30秒
    
    # 6. 清理蓝色环境（可选，保留一段时间以便快速回滚）
    # print("Cleaning up blue environment...")
    # subprocess.run([
    #     "kubectl", "delete", "deployment", "llm-inference-blue"
    # ], check=True)
    
    print("Blue-green deployment completed successfully!")

if __name__ == "__main__":
    blue_green_deployment()

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import random
import hashlib
import redis
import time
import threading

app = FastAPI()

# 连接Redis用于存储实验数据
r = redis.Redis(host='localhost', port=6379, db=0)

# 实验配置
EXPERIMENTS = {
    "llm_model_variant": {
        "variants": {
            "A": {"weight": 0.4, "model_version": "v1.0"},
            "B": {"weight": 0.4, "model_version": "v2.0"},
            "control": {"weight": 0.2, "model_version": "v0.5"}
        },
        "start_time": time.time(),
        "end_time": time.time() + 7 * 24 * 3600  # 7天后结束
    }
}

# 根据用户ID分配实验变体
def assign_variant(user_id, experiment_name):
    experiment = EXPERIMENTS.get(experiment_name)
    if not experiment:
        return None
    
    # 检查实验是否在运行中
    if time.time() < experiment["start_time"] or time.time() > experiment["end_time"]:
        return None
    
    # 为用户生成一个一致的哈希值
    hash_key = f"{experiment_name}:{user_id}"
    hash_value = hashlib.md5(hash_key.encode()).hexdigest()
    
    # 使用哈希值决定变体
    rng = random.Random(int(hash_value, 16))
    
    # 根据权重分配变体
    total_weight = sum(v["weight"] for v in experiment["variants"].values())
    r_value = rng.uniform(0, total_weight)
    
    current_weight = 0
    for variant, config in experiment["variants"].items():
        current_weight += config["weight"]
        if r_value <= current_weight:
            # 记录用户分配
            r.hset(f"user_experiment:{user_id}", experiment_name, variant)
            return variant
    
    return "control"  # 默认返回控制组

# 记录实验指标
def record_metric(user_id, experiment_name, variant, metric_name, metric_value):
    # 记录个人指标
    user_key = f"user_metric:{user_id}:{experiment_name}:{variant}:{metric_name}"
    r.lpush(user_key, metric_value)
    r.ltrim(user_key, 0, 99)  # 保留最近的100个值
    
    # 记录聚合指标
    aggregate_key = f"aggregate_metric:{experiment_name}:{variant}:{metric_name}"
    r.lpush(aggregate_key, metric_value)
    r.ltrim(aggregate_key, 0, 9999)  # 保留最近的10000个值

# 模拟模型推理函数
def model_inference(text, model_version):
    # 在实际应用中，这里会调用不同版本的模型
    print(f"Using model version: {model_version}")
    time.sleep(0.5)  # 模拟推理时间
    return f"Response from model {model_version}: {text}"

@app.post("/inference")
async def inference(request: Request):
    data = await request.json()
    text = data.get("text", "")
    user_id = data.get("user_id", "anonymous")
    
    # 为用户分配实验变体
    variant = assign_variant(user_id, "llm_model_variant")
    if not variant:
        variant = "control"
    
    # 获取对应的模型版本
    model_version = EXPERIMENTS["llm_model_variant"]["variants"][variant]["model_version"]
    
    # 执行推理
    start_time = time.time()
    response = model_inference(text, model_version)
    inference_time = (time.time() - start_time) * 1000  # 转换为毫秒
    
    # 记录性能指标
    record_metric(user_id, "llm_model_variant", variant, "inference_time", inference_time)
    
    return JSONResponse({
        "response": response,
        "experiment_variant": variant,
        "model_version": model_version
    })

@app.post("/feedback")
async def feedback(request: Request):
    data = await request.json()
    user_id = data.get("user_id", "anonymous")
    rating = data.get("rating", 0)  # 1-5的评分
    
    # 获取用户的实验变体
    variant = r.hget(f"user_experiment:{user_id}", "llm_model_variant")
    if variant:
        variant = variant.decode()
        # 记录用户反馈
        record_metric(user_id, "llm_model_variant", variant, "user_rating", rating)
    
    return {"status": "success"}

@app.get("/experiment_results")
async def experiment_results():
    experiment_name = "llm_model_variant"
    results = {}
    
    # 获取每个变体的平均指标
    for variant in EXPERIMENTS[experiment_name]["variants"].keys():
        # 推理时间
        inference_times = r.lrange(f"aggregate_metric:{experiment_name}:{variant}:inference_time", 0, -1)
        avg_inference_time = sum(float(t) for t in inference_times) / len(inference_times) if inference_times else 0
        
        # 用户评分
        ratings = r.lrange(f"aggregate_metric:{experiment_name}:{variant}:user_rating", 0, -1)
        avg_rating = sum(int(r) for r in ratings) / len(ratings) if ratings else 0
        
        results[variant] = {
            "avg_inference_time_ms": round(avg_inference_time, 2),
            "avg_user_rating": round(avg_rating, 2),
            "sample_size": len(inference_times)
        }
    
    return results

# 启动一个后台线程来定期检查实验结果
def monitor_experiment():
    while True:
        # 检查是否有明显的优胜者
        results = experiment_results()
        
        # 这里可以添加自动停止实验的逻辑
        # 例如，如果某个变体的性能明显优于其他变体，就提前结束实验
        
        print("Experiment monitoring update:")
        for variant, metrics in results.items():
            print(f"Variant {variant}: {metrics}")
        
        time.sleep(3600)  # 每小时检查一次

# 启动监控线程
threading.Thread(target=monitor_experiment, daemon=True).start()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

import random
import time
import subprocess
import logging

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger('chaos_engineering')

class ChaosEngine:
    def __init__(self, config=None):
        self.config = config or {
            "enabled": True,
            "failure_probability": 0.05,  # 5%的概率注入故障
            "max_failures_per_run": 3,
            "allowed_failure_types": ["pod_kill", "network_latency", "cpu_stress", "memory_stress"],
            "time_window": {"start": "08:00", "end": "20:00"},  # 只在工作时间进行
            "excluded_days": ["Saturday", "Sunday"]  # 排除周末
        }
    
    def is_execution_window(self):
        """检查当前时间是否在允许的执行窗口内"""
        import datetime
        now = datetime.datetime.now()
        
        # 检查是否是排除的日期
        if now.strftime("%A") in self.config["excluded_days"]:
            return False
        
        # 检查时间范围
        current_time = now.time()
        start_time = datetime.datetime.strptime(self.config["time_window"]["start"], "%H:%M").time()
        end_time = datetime.datetime.strptime(self.config["time_window"]["end"], "%H:%M").time()
        
        return start_time <= current_time <= end_time
    
    def inject_pod_failure(self):
        """随机终止一个Pod"""
        try:
            # 获取所有运行中的pod
            result = subprocess.run(
                ["kubectl", "get", "pods", "-l", "app=llm-service", "-o", "jsonpath={.items[*].metadata.name}"],
                capture_output=True, text=True, check=True
            )
            
            pods = result.stdout.split()
            if not pods:
                logger.warning("No pods found to inject failure")
                return False
            
            # 随机选择一个pod
            target_pod = random.choice(pods)
            logger.info(f"Injecting pod failure: killing {target_pod}")
            
            # 终止pod
            subprocess.run(
                ["kubectl", "delete", "pod", target_pod],
                capture_output=True, text=True, check=True
            )
            
            logger.info(f"Successfully killed pod {target_pod}")
            return True
        except Exception as e:
            logger.error(f"Error injecting pod failure: {e}")
            return False
    
    def inject_network_latency(self):
        """为Pod添加网络延迟"""
        try:
            # 获取所有运行中的pod
            result = subprocess.run(
                ["kubectl", "get", "pods", "-l", "app=llm-service", "-o", "jsonpath={.items[*].metadata.name}"],
                capture_output=True, text=True, check=True
            )
            
            pods = result.stdout.split()
            if not pods:
                logger.warning("No pods found to inject network latency")
                return False
            
            # 随机选择一个pod
            target_pod = random.choice(pods)
            latency_ms = random.randint(100, 500)  # 100-500毫秒延迟
            
            logger.info(f"Injecting network latency: adding {latency_ms}ms to {target_pod}")
            
            # 使用tc命令添加延迟（需要pod有足够权限）
            # 注意：这需要在pod内执行，这里只是示例
            command = f"kubectl exec {target_pod} -- tc qdisc add dev eth0 root netem delay {latency_ms}ms"
            subprocess.run(command, shell=True, capture_output=True, text=True, check=True)
            
            # 设置延迟30秒后恢复
            def restore_network():
                time.sleep(30)
                restore_command = f"kubectl exec {target_pod} -- tc qdisc del dev eth0 root netem"
                try:
                    subprocess.run(restore_command, shell=True, capture_output=True, text=True, check=True)
                    logger.info(f"Network latency restored for pod {target_pod}")
                except Exception as e:
                    logger.error(f"Error restoring network: {e}")
            
            import threading
            threading.Thread(target=restore_network).start()
            
            return True
        except Exception as e:
            logger.error(f"Error injecting network latency: {e}")
            return False
    
    def inject_cpu_stress(self):
        """对Pod进行CPU压力测试"""
        try:
            # 获取所有运行中的pod
            result = subprocess.run(
                ["kubectl", "get", "pods", "-l", "app=llm-service", "-o", "jsonpath={.items[*].metadata.name}"],
                capture_output=True, text=True, check=True
            )
            
            pods = result.stdout.split()
            if not pods:
                logger.warning("No pods found to inject CPU stress")
                return False
            
            # 随机选择一个pod
            target_pod = random.choice(pods)
            
            logger.info(f"Injecting CPU stress to pod {target_pod}")
            
            # 在pod内运行stress命令（需要pod中有stress工具）
            # 注意：这只是示例，实际应用中需要确保pod中有stress工具
            command = f"kubectl exec {target_pod} -- stress --cpu 4 --timeout 30s"
            subprocess.Popen(command, shell=True)
            
            return True
        except Exception as e:
            logger.error(f"Error injecting CPU stress: {e}")
            return False
    
    def inject_memory_stress(self):
        """对Pod进行内存压力测试"""
        try:
            # 获取所有运行中的pod
            result = subprocess.run(
                ["kubectl", "get", "pods", "-l", "app=llm-service", "-o", "jsonpath={.items[*].metadata.name}"],
                capture_output=True, text=True, check=True
            )
            
            pods = result.stdout.split()
            if not pods:
                logger.warning("No pods found to inject memory stress")
                return False
            
            # 随机选择一个pod
            target_pod = random.choice(pods)
            
            logger.info(f"Injecting memory stress to pod {target_pod}")
            
            # 在pod内运行stress命令，分配2GB内存
            command = f"kubectl exec {target_pod} -- stress --vm 2 --vm-bytes 1G --timeout 30s"
            subprocess.Popen(command, shell=True)
            
            return True
        except Exception as e:
            logger.error(f"Error injecting memory stress: {e}")
            return False
    
    def run_experiment(self):
        """运行混沌实验"""
        if not self.config["enabled"]:
            logger.info("Chaos engineering is disabled")
            return
        
        if not self.is_execution_window():
            logger.info("Current time is outside the execution window")
            return
        
        # 根据配置的概率决定是否执行故障注入
        if random.random() > self.config["failure_probability"]:
            return
        
        logger.info("Starting chaos experiment")
        
        # 随机选择故障类型
        failure_type = random.choice(self.config["allowed_failure_types"])
        
        # 执行对应的故障注入
        if failure_type == "pod_kill":
            self.inject_pod_failure()
        elif failure_type == "network_latency":
            self.inject_network_latency()
        elif failure_type == "cpu_stress":
            self.inject_cpu_stress()
        elif failure_type == "memory_stress":
            self.inject_memory_stress()
        
        logger.info("Chaos experiment completed")

# 运行示例
if __name__ == "__main__":
    chaos = ChaosEngine()
    
    # 在实际应用中，这应该作为一个定时任务运行
    while True:
        chaos.run_experiment()
        time.sleep(60)  # 每分钟检查一次是否执行实验

import opentelemetry
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from fastapi import FastAPI, Depends
import time
import random

# 配置OpenTelemetry
resource = Resource(attributes={
    SERVICE_NAME: "llm-inference-service"
})

provider = TracerProvider(resource=resource)

# 配置Jaeger导出器
jaeger_exporter = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831,
)

# 添加处理器
processor = BatchSpanProcessor(jaeger_exporter)
provider.add_span_processor(processor)

# 设置全局追踪提供程序
trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)

app = FastAPI()

# 模拟模型加载和初始化
def load_model(model_name):
    with tracer.start_as_current_span("model_loading") as span:
        span.set_attribute("model.name", model_name)
        # 模拟加载时间
        time.sleep(0.2)
        span.set_attribute("model.loaded", True)
        return f"model_{model_name}"

# 模拟分词处理
def tokenize(text):
    with tracer.start_as_current_span("tokenization") as span:
        span.set_attribute("input.length", len(text))
        # 模拟分词时间
        time.sleep(0.1)
        tokens = text.split()
        span.set_attribute("token.count", len(tokens))
        return tokens

# 模拟GPU内存分配
def allocate_gpu_memory(size_mb):
    with tracer.start_as_current_span("gpu_memory_allocation") as span:
        span.set_attribute("memory.size_mb", size_mb)
        # 模拟分配时间
        time.sleep(0.05)
        span.set_attribute("allocation.success", True)
        return f"gpu_memory_block_{random.randint(1, 1000)}"

# 模拟推理过程
def inference_step(input_tokens, model):
    with tracer.start_as_current_span("inference") as span:
        span.set_attribute("model", model)
        span.set_attribute("token.count", len(input_tokens))
        
        # 模拟注意力计算
        with tracer.start_as_current_span("attention_computation"):
            time.sleep(0.2)
            
        # 模拟前馈网络计算
        with tracer.start_as_current_span("feed_forward_computation"):
            time.sleep(0.3)
        
        # 模拟输出处理
        with tracer.start_as_current_span("output_processing"):
            time.sleep(0.1)
        
        output = f"Processed {len(input_tokens)} tokens with {model}"
        span.set_attribute("output.length", len(output))
        return output

# 主推理端点
@app.post("/inference")
async def inference_endpoint(request: dict):
    with tracer.start_as_current_span("inference_request") as span:
        text = request.get("text", "")
        model_name = request.get("model", "default")
        
        span.set_attribute("request.text_length", len(text))
        span.set_attribute("request.model", model_name)
        
        try:
            # 步骤1: 加载模型
            model = load_model(model_name)
            
            # 步骤2: 分词
            tokens = tokenize(text)
            
            # 步骤3: 分配GPU内存
            memory_block = allocate_gpu_memory(len(tokens) * 0.5)  # 假设每个token需要0.5MB内存
            
            # 步骤4: 执行推理
            result = inference_step(tokens, model)
            
            span.set_attribute("request.success", True)
            return {"result": result, "model": model_name}
            
        except Exception as e:
            span.set_attribute("request.success", False)
            span.record_exception(e)
            raise

# 健康检查端点
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# Kubernetes集群架构示例
apiVersion: v1
kind: Namespace
metadata:
  name: llm-service
---
# 服务部署（3个副本）
apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-inference
  namespace: llm-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: llm-inference
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 1
  template:
    metadata:
      labels:
        app: llm-inference
    spec:
      affinity:
        podAntiAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 100
            podAffinityTerm:
              labelSelector:
                matchExpressions:
                - key: app
                  operator: In
                  values:
                  - llm-inference
              topologyKey: "kubernetes.io/hostname"
      containers:
      - name: llm-inference
        image: llm-inference:latest
        resources:
          requests:
            memory: "16Gi"
            cpu: "4"
            nvidia.com/gpu: 1
          limits:
            memory: "24Gi"
            cpu: "8"
            nvidia.com/gpu: 1
        ports:
        - containerPort: 8000
        livenessProbe:
          httpGet:
            path: /live
            port: 8000
          initialDelaySeconds: 60
          periodSeconds: 30
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 15
        volumeMounts:
        - name: model-cache
          mountPath: /app/model_cache
      volumes:
      - name: model-cache
        emptyDir: {}
---
# 服务定义
apiVersion: v1
kind: Service
metadata:
  name: llm-inference-service
  namespace: llm-service
spec:
  selector:
    app: llm-inference
  ports:
  - port: 80
    targetPort: 8000
  type: ClusterIP
---
# 水平Pod自动缩放器
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: llm-inference-hpa
  namespace: llm-service
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: llm-inference
  minReplicas: 3
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80
---
# Redis缓存（用于KV缓存和请求队列）
apiVersion: apps/v1
kind: Deployment
metadata:
  name: redis
  namespace: llm-service
spec:
  replicas: 1
  selector:
    matchLabels:
      app: redis
  template:
    metadata:
      labels:
        app: redis
    spec:
      containers:
      - name: redis
        image: redis:6.2
        ports:
        - containerPort: 6379
        volumeMounts:
        - name: redis-data
          mountPath: /data
      volumes:
      - name: redis-data
        persistentVolumeClaim:
          claimName: redis-pvc
---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: redis-pvc
  namespace: llm-service
spec:
  accessModes:
    - ReadWriteOnce
  resources:
    requests:
      storage: 50Gi
---
apiVersion: v1
kind: Service
metadata:
  name: redis
  namespace: llm-service
spec:
  selector:
    app: redis
  ports:
  - port: 6379
    targetPort: 6379