AI辅助微服务开发与运维

# 分层异常检测示例（Python + Prometheus Client）
# Prometheus 3.x 兼容
import numpy as np
from prometheus_client import CollectorRegistry, Gauge
from datetime import datetime, timedelta
 
class LayeredAnomalyDetector:
    """分层异常检测器：快速检测 + 精确确认"""
 
    def __init__(self, metric_name: str):
        self.metric_name = metric_name
        self.history_window = 7 * 24 * 3600  # 7 天历史数据
 
    def layer1_statistical_check(self, current: float,
                                  baseline: np.ndarray) -> dict:
        """第一层：统计快速检测（低延迟）"""
        mean = np.mean(baseline)
        std = np.std(baseline)
        z_score = (current - mean) / std if std > 0 else 0
 
        return {
            "is_anomaly": abs(z_score) > 3,
            "z_score": z_score,
            "confidence": "low",  # 统计检测置信度较低
            "method": "z_score"
        }
 
    def layer2_seasonal_check(self, current: float,
                               history: np.ndarray,
                               period: int = 86400) -> dict:
        """第二层：周期性检测（中延迟）"""
        # 取同周期历史数据
        same_period_values = history[-period:]
        period_mean = np.mean(same_period_values)
        period_std = np.std(same_period_values)
        deviation = abs(current - period_mean) / period_std \
                    if period_std > 0 else 0
 
        return {
            "is_anomaly": deviation > 2.5,
            "deviation": deviation,
            "confidence": "medium",
            "method": "seasonal_decomposition"
        }
 
    def layer3_ml_confirm(self, features: np.ndarray,
                          model) -> dict:
        """第三层：ML 模型确认（高延迟、高准确率）"""
        prediction = model.predict(features.reshape(1, -1))
        is_anomaly = prediction[0] == -1  # Isolation Forest
 
        return {
            "is_anomaly": bool(is_anomaly),
            "confidence": "high",
            "method": "isolation_forest"
        }

// order.proto - 订单服务接口定义
syntax = "proto3";
package order.v1;
 
service OrderService {
  // AI 自动补全：基于已有字段和业务语义
  // 生成 CreateOrder、GetOrder、ListOrders 等接口
  rpc CreateOrder(CreateOrderRequest) returns (CreateOrderResponse);
  rpc GetOrder(GetOrderRequest) returns (GetOrderResponse);
  rpc ListOrders(ListOrdersRequest) returns (ListOrdersResponse);
  rpc UpdateOrderStatus(UpdateOrderStatusRequest) returns (UpdateOrderStatusResponse);
}

# .cursorrules - 项目级 AI 编码规则
 
## 项目架构
- 采用 Clean Architecture 分层：handler / service / repository / model
- 使用 Kratos 框架（Go microservice framework）
- 配置管理使用 Viper，配置文件为 YAML
- 数据库访问使用 GORM，禁止裸 SQL
 
## 编码规范
- 错误处理：使用 errors.Wrap 包装错误上下文
- 日志：使用 logrus，禁止 fmt.Println
- 命名：遵循 Go 官方命名规范，接口不加 I 前缀
- 并发：优先使用 errgroup 管理 goroutine 生命周期
 
## 微服务特有规则
- 所有对外接口必须有超时控制（context.WithTimeout）
- 服务间调用必须实现重试 + 熔断（使用 kratos middleware）
- 每个接口必须有 OpenTelemetry 的 Trace 和 Metrics 埋点

# AI 生成的 Kubernetes Deployment（Kubernetes 1.30+）
apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-service
  namespace: production
  labels:
    app: order-service
    version: v2.1.0
spec:
  replicas: 3
  selector:
    matchLabels:
      app: order-service
  template:
    metadata:
      labels:
        app: order-service
        version: v2.1.0
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "9090"
        # OpenTelemetry 注入
        instrumentation.opentelemetry.io/inject-java: "true"
    spec:
      containers:
        - name: order-service
          image: registry.example.com/order-service:v2.1.0
          ports:
            - containerPort: 8080
              name: http
            - containerPort: 9090
              name: metrics
          resources:
            requests:
              cpu: "500m"
              memory: "512Mi"
            limits:
              cpu: "2000m"
              memory: "2Gi"
          env:
            - name: OTEL_SERVICE_NAME
              value: "order-service"
            - name: OTEL_EXPORTER_OTLP_ENDPOINT
              value: "http://otel-collector:4317"
          livenessProbe:
            httpGet:
              path: /healthz
              port: 8080
            initialDelaySeconds: 15
            periodSeconds: 10
          readinessProbe:
            httpGet:
              path: /ready
              port: 8080
            initialDelaySeconds: 5
            periodSeconds: 5

// order_service.go - 订单服务核心逻辑
func (s *OrderService) CreateOrder(ctx context.Context,
    req *pb.CreateOrderRequest) (*pb.CreateOrderResponse, error) {
    // 参数校验
    if req.UserId == "" || len(req.Items) == 0 {
        return nil, errors.New("invalid request: user_id and items required")
    }
 
    // 库存检查（调用库存服务）
    for _, item := range req.Items {
        available, err := s.inventoryClient.CheckStock(ctx, item.SkuId)
        if err != nil {
            return nil, errors.Wrap(err, "check stock failed")
        }
        if !available {
            return nil, fmt.Errorf("sku %s out of stock", item.SkuId)
        }
    }
 
    // 创建订单
    order := &Order{
        OrderId:   uuid.New().String(),
        UserId:    req.UserId,
        Items:     req.Items,
        Status:    OrderStatus_CREATED,
        CreatedAt: time.Now(),
    }
 
    if err := s.repo.Save(ctx, order); err != nil {
        return nil, errors.Wrap(err, "save order failed")
    }
 
    return &pb.CreateOrderResponse{OrderId: order.OrderId}, nil
}

SRE: "订单服务今天下午 2 点开始 P99 延迟升高，帮忙排查"
 
AI Agent 执行步骤：
1. 查询 order-service 14:00-15:00 的 P99 延迟趋势 → 确认异常
2. 检查该时段 error rate → 发现 5xx 从 0.1% 升至 3.5%
3. 分析 Trace 数据 → 定位到 payment-service 调用延迟异常
4. 检查 payment-service 指标 → 发现 DB 连接池使用率达 95%
5. 查询 slow query log → 发现一条新增的全表扫描 SQL
6. 根因结论：新上线的代码引入了缺少索引的查询条件
 
修复建议：
1. 紧急：为该查询条件添加数据库索引
2. 临时：对 payment-service 限流，保护 DB 连接池
3. 长期：SQL 上线前增加 Explain 审查环节

# 混合预测模型示例
# 依赖：prophet>=1.1, tensorflow>=2.15
import numpy as np
import pandas as pd
from prophet import Prophet
 
class HybridCapacityForecaster:
    """Prophet + LSTM 混合容量预测器"""
 
    def __init__(self, service_name: str,
                 forecast_horizon: str = "24h"):
        self.service_name = service_name
        self.forecast_horizon = forecast_horizon
        self.prophet_model = Prophet(
            yearly_seasonality=True,
            weekly_seasonality=True,
            daily_seasonality=True,
            changepoint_prior_scale=0.05
        )
 
    def train(self, metrics_df: pd.DataFrame):
        """
        metrics_df: columns = [ds, y]
        ds: datetime, y: QPS 或资源利用率
        """
        # 第一层：Prophet 捕捉周期性
        self.prophet_model.fit(metrics_df)
 
        # 计算残差
        forecast = self.prophet_model.predict(metrics_df)
        residuals = metrics_df['y'].values - \
                    forecast['yhat'].values
 
        # 第二层：LSTM 捕捉残差中的非线性模式
        # 此处简化，实际需构建序列窗口
        self.residual_mean = np.mean(residuals)
        self.residual_std = np.std(residuals)
 
    def predict(self, future_df: pd.DataFrame) -> pd.DataFrame:
        """预测未来流量"""
        # Prophet 周期性预测
        prophet_forecast = self.prophet_model.predict(future_df)
 
        # 合并预测结果
        result = prophet_forecast[['ds', 'yhat']].copy()
        result.columns = ['timestamp', 'predicted_qps']
 
        # 添加置信区间
        result['lower_bound'] = prophet_forecast['yhat_lower']
        result['upper_bound'] = prophet_forecast['yhat_upper']
 
        # 容量建议：基于预测值 + 安全余量
        safety_margin = 1.2  # 20% 安全余量
        result['recommended_instances'] = np.ceil(
            result['predicted_qps'] * safety_margin /
            self.single_instance_capacity
        )
 
        return result

# Predictive HPA 配置示例（基于 KEDA + 自定义预测指标）
# Kubernetes 1.30+, KEDA 2.14+
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: order-service-predictive
  namespace: production
spec:
  scaleTargetRef:
    name: order-service
  minReplicaCount: 3
  maxReplicaCount: 50
  cooldownPeriod: 60
  triggers:
    # 实时指标触发器（反应式）
    - type: prometheus
      metadata:
        serverAddress: http://prometheus:9090
        metricName: http_requests_per_second
        threshold: "1000"
        query: >
          sum(rate(http_requests_total{
            service="order-service"
          }[1m]))
    # 预测指标触发器（预测式）
    - type: prometheus
      metadata:
        serverAddress: http://prometheus:9090
        metricName: predicted_qps_30m_ahead
        threshold: "800"
        query: >
          predicted_qps{
            service="order-service",
            horizon="30m"
          }

原始日志:
2025-01-15 14:02:03 [ERROR] OrderService - Failed to process order #12345: connection timeout to payment-service:8080
2025-01-15 14:02:04 [ERROR] OrderService - Failed to process order #12346: connection timeout to payment-service:8080
2025-01-15 14:02:05 [ERROR] OrderService - Failed to process order #12347: connection timeout to payment-service:8080
 
日志模板（Drain 算法输出）:
[ERROR] OrderService - Failed to process order <*>, connection timeout to payment-service:8080
 
统计：3 次 / 3秒（异常突增）

# 基于 LLM 的日志异常分析示例
# Python 3.11+
import json
from dataclasses import dataclass
 
@dataclass
class LogAnomaly:
    timestamp: str
    service: str
    log_template: str
    anomaly_type: str
    severity: str
    root_cause_hypothesis: str
    suggested_action: str
 
class LLMLogAnalyzer:
    """LLM 驱动的日志异常分析器"""
 
    ANALYSIS_PROMPT = """
    你是一位微服务运维专家。请分析以下异常日志模式：
 
    服务：{service}
    时间窗口：{time_window}
    异常日志模式：{log_pattern}
    出现频率：{frequency}（基线：{baseline}）
    关联指标变化：{related_metrics}
 
    请输出 JSON 格式：
    {{
        "anomaly_type": "网络超时/资源耗尽/代码缺陷/配置错误/安全事件",
        "severity": "P0/P1/P2/P3",
        "root_cause_hypothesis": "根因假设",
        "confidence": 0.0-1.0,
        "suggested_actions": ["动作1", "动作2"],
        "affected_services": ["服务列表"]
    }}
    """
 
    def analyze(self, anomaly_context: dict) -> LogAnomaly:
        prompt = self.ANALYSIS_PROMPT.format(**anomaly_context)
        # 调用 LLM API（Claude / GPT-4 / 本地模型）
        response = self._call_llm(prompt)
        result = json.loads(response)
 
        return LogAnomaly(
            timestamp=anomaly_context["time_window"],
            service=anomaly_context["service"],
            log_template=anomaly_context["log_pattern"],
            anomaly_type=result["anomaly_type"],
            severity=result["severity"],
            root_cause_hypothesis=result["root_cause_hypothesis"],
            suggested_action="; ".join(result["suggested_actions"])
        )

正常请求（P50）: Gateway[2ms] → Order[15ms] → Payment[30ms] → DB[10ms] = 57ms
异常请求（P99）: Gateway[2ms] → Order[45ms] → Payment[150ms] → DB[120ms] = 317ms
 
差异分析：
- DB 延迟增长 12x → 根因候选 #1
- Payment 延迟增长 5x → 由 DB 延迟传导
- Order 延迟增长 3x → 由 Payment 延迟传导

# 流量基线建模示例
# Python 3.11+, scikit-learn >= 1.4
import numpy as np
from sklearn.ensemble import IsolationForest
from dataclasses import dataclass
 
@dataclass
class TrafficFeatures:
    qps: float
    avg_request_size: float
    error_rate: float
    unique_source_count: int
    p99_latency: float
    ratio_post_get: float  # POST/GET 请求比例
 
class TrafficAnomalyDetector:
    """微服务流量异常检测器"""
 
    def __init__(self, contamination: float = 0.01):
        self.model = IsolationForest(
            n_estimators=100,
            contamination=contamination,
            random_state=42
        )
        self.is_trained = False
 
    def train(self, normal_traffic: list[TrafficFeatures]):
        """基于正常流量训练基线模型"""
        X = np.array([
            [f.qps, f.avg_request_size, f.error_rate,
             f.unique_source_count, f.p99_latency,
             f.ratio_post_get]
            for f in normal_traffic
        ])
        self.model.fit(X)
        self.is_trained = True
 
    def detect(self, traffic: TrafficFeatures) -> dict:
        """检测流量异常"""
        if not self.is_trained:
            raise RuntimeError("Model not trained")
 
        X = np.array([[
            traffic.qps, traffic.avg_request_size,
            traffic.error_rate, traffic.unique_source_count,
            traffic.p99_latency, traffic.ratio_post_get
        ]])
 
        prediction = self.model.predict(X)[0]    # -1 = 异常
        score = self.model.score_samples(X)[0]   # 异常分数
 
        return {
            "is_anomaly": prediction == -1,
            "anomaly_score": float(score),
            "severity": self._calc_severity(score),
            "details": self._explain_anomaly(traffic)
        }
 
    def _calc_severity(self, score: float) -> str:
        if score < -0.5:
            return "CRITICAL"
        elif score < -0.3:
            return "HIGH"
        elif score < -0.1:
            return "MEDIUM"
        return "LOW"
 
    def _explain_anomaly(self, traffic: TrafficFeatures) -> str:
        explanations = []
        if traffic.error_rate > 0.1:
            explanations.append("错误率异常升高")
        if traffic.ratio_post_get > 5.0:
            explanations.append("POST 请求比例异常（疑似暴力提交）")
        if traffic.unique_source_count < 2 and traffic.qps > 1000:
            explanations.append("单一来源高频请求（疑似 DDoS）")
        return "; ".join(explanations) or "综合特征异常"

// 漏洞代码（SQL 注入风险）
public User findUser(String username) {
    String sql = "SELECT * FROM users WHERE username = '"
                 + username + "'";
    return jdbcTemplate.queryForObject(sql, userRowMapper);
}
 
// AI 自动生成的修复代码（参数化查询）
public User findUser(String username) {
    String sql = "SELECT * FROM users WHERE username = ?";
    return jdbcTemplate.queryForObject(
        sql, userRowMapper, username);
}

# AI 生成的 Istio AuthorizationPolicy
# Istio 1.21+
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: order-service-access
  namespace: production
spec:
  selector:
    matchLabels:
      app: order-service
  rules:
    # 仅允许以下服务访问 order-service
    - from:
        - source:
            principals:
              - "cluster.local/ns/production/sa/api-gateway"
              - "cluster.local/ns/production/sa/user-service"
      to:
        - operation:
            methods: ["GET", "POST"]
            paths: ["/api/v1/orders/*"]
    # 管理接口仅限运维命名空间访问
    - from:
        - source:
            namespaces: ["ops"]
      to:
        - operation:
            methods: ["GET"]
            paths: ["/admin/*"]