Agent 版本控制、灰度与回滚机制

┌──────────────────────────────────────────────────────────────┐
│                  金丝雀发布流程状态机                        │
├──────────────────────────────────────────────────────────────┤
│                                                              │
│  [准备阶段]                                                  │
│      │                                                       │
│      ▼                                                       │
│  ┌─────────┐                                                │
│  │ 构建镜像 │                                                │
│  └────┬────┘                                                │
│       │                                                      │
│       ▼                                                      │
│  ┌─────────┐                                                │
│  │ 部署金丝雀│ ← 部署 1-5% 流量                              │
│  └────┬────┘                                                │
│       │                                                      │
│       ▼                                                      │
│  [观察阶段]                                                  │
│       │                                                      │
│       ▼                                                      │
│  ┌─────────┐     指标正常      ┌─────────┐                 │
│  │ 监控指标│ ────────────────→ │ 扩大比例│                 │
│  └────┬────┘                   └────┬────┘                 │
│       │                              │                       │
│       │ 指标异常                     │                       │
│       ▼                              │                       │
│  ┌─────────┐                        │                       │
│  │ 触发回滚│                        │                       │
│  └────┬────┘                        │                       │
│       │                              │                       │
│       └──────────┬──────────────────┘                       │
│                  │                                          │
│                  ▼                                          │
│         比例达到 100%？                                     │
│                  │                                          │
│           ┌──────┴──────┐                                  │
│           │             │                                  │
│          是            否                                  │
│           │             │                                  │
│           ▼             └──────────┐                       │
│  ┌─────────┐                      │                       │
│  │ 全量发布│ ←────────────────────┘                       │
│  └────┬────┘                                                │
│       │                                                      │
│       ▼                                                      │
│  ┌─────────┐                                                │
│  │ 清理旧版│                                                │
│  └────┬────┘                                                │
│       │                                                      │
│       ▼                                                      │
│  [完成阶段]                                                  │
│                                                              │
└──────────────────────────────────────────────────────────────┘

状态转换规则：
1. 准备 → 部署：构建完成，开始部署金丝雀实例
2. 部署 → 观察：部署完成，开始监控指标
3. 观察 → 扩大：指标正常（错误率<1%，延迟<200ms），扩大流量比例
4. 观察 → 回滚：指标异常（错误率>5%，延迟>1s），触发回滚
5. 扩大 → 观察：比例扩大后，继续监控
6. 扩大 → 完成：比例达到 100%，发布完成
                        

import hashlib
from typing import Dict, List

class CanaryTrafficSplitter:
    """金丝雀流量分割器"""
    
    def __init__(self, canary_percentage: float = 0.05):
        """
        初始化流量分割器
        
        Args:
            canary_percentage: 金丝雀流量比例 (0.0-1.0)
        """
        self.canary_percentage = canary_percentage
    
    def should_route_to_canary(
        self, 
        request_id: str,
        user_id: str = None,
        headers: Dict[str, str] = None
    ) -> bool:
        """
        判断请求是否路由到金丝雀版本
        
        使用一致性哈希保证同一用户的请求总是路由到同一版本
        
        Args:
            request_id: 请求 ID
            user_id: 用户 ID（可选）
            headers: 请求头（可选）
        
        Returns:
            bool: True 表示路由到金丝雀版本
        """
        # 生成一致性哈希键
        hash_key = self._generate_hash_key(request_id, user_id, headers)
        
        # 计算哈希值 (0-100)
        hash_value = self._consistent_hash(hash_key) % 100
        
        # 判断是否落在金丝雀区间
        threshold = int(self.canary_percentage * 100)
        return hash_value < threshold
    
    def _generate_hash_key(
        self, 
        request_id: str,
        user_id: str = None,
        headers: Dict[str, str] = None
    ) -> str:
        """生成一致性哈希键"""
        
        # 优先使用 user_id 保证用户粘性
        if user_id:
            return f"user:{user_id}"
        
        # 其次使用特定 header（如设备 ID）
        if headers and 'X-Device-ID' in headers:
            return f"device:{headers['X-Device-ID']}"
        
        # 最后使用 request_id
        return f"request:{request_id}"
    
    def _consistent_hash(self, key: str) -> int:
        """计算一致性哈希值"""
        hash_obj = hashlib.md5(key.encode('utf-8'))
        hash_hex = hash_obj.hexdigest()
        # 取前 8 位转换为整数
        return int(hash_hex[:8], 16)
    
    def update_percentage(self, new_percentage: float):
        """更新金丝雀流量比例"""
        if not 0.0 <= new_percentage <= 1.0:
            raise ValueError("Percentage must be between 0.0 and 1.0")
        self.canary_percentage = new_percentage


# 使用示例
splitter = CanaryTrafficSplitter(canary_percentage=0.05)  # 5% 金丝雀流量

# 模拟请求路由
for i in range(1000):
    request_id = f"req-{i}"
    user_id = f"user-{i % 100}"  # 100 个用户
    
    if splitter.should_route_to_canary(request_id, user_id):
        # 路由到金丝雀版本 (v2)
        route_to = "canary-v2"
    else:
        # 路由到稳定版本 (v1)
        route_to = "stable-v1"

# 渐进式扩大流量
stages = [0.05, 0.10, 0.25, 0.50, 0.75, 1.0]
for percentage in stages:
    splitter.update_percentage(percentage)
    print(f"金丝雀流量比例：{percentage * 100:.0f}%")
    # 在每个阶段观察指标，决定是否继续扩大
                        

class CanaryDecisionEngine:
    """金丝雀发布决策引擎"""
    
    def __init__(self):
        self.thresholds = {
            'error_rate': {'warning': 0.01, 'critical': 0.05},
            'p95_latency': {'warning': 200, 'critical': 1000},  # ms
            'throughput_drop': {'warning': 0.10, 'critical': 0.30},
            'business_metric_drop': {'warning': 0.05, 'critical': 0.15}
        }
    
    def evaluate_canary_health(
        self, 
        canary_metrics: Dict,
        baseline_metrics: Dict
    ) -> Dict:
        """
        评估金丝雀版本健康度
        
        Returns:
            Dict: {
                'status': 'healthy' | 'warning' | 'critical',
                'score': 0-100,
                'issues': List[str],
                'recommendation': 'proceed' | 'hold' | 'rollback'
            }
        """
        issues = []
        score = 100
        
        # 1. 错误率检查
        error_rate = canary_metrics.get('error_rate', 0)
        if error_rate > self.thresholds['error_rate']['critical']:
            issues.append(f"错误率过高：{error_rate:.2%}")
            score -= 40
        elif error_rate > self.thresholds['error_rate']['warning']:
            issues.append(f"错误率警告：{error_rate:.2%}")
            score -= 20
        
        # 2. 延迟检查
        p95_latency = canary_metrics.get('p95_latency', 0)
        if p95_latency > self.thresholds['p95_latency']['critical']:
            issues.append(f"P95 延迟过高：{p95_latency:.0f}ms")
            score -= 40
        elif p95_latency > self.thresholds['p95_latency']['warning']:
            issues.append(f"P95 延迟警告：{p95_latency:.0f}ms")
            score -= 20
        
        # 3. 吞吐量检查
        throughput_drop = self._calculate_drop(
            canary_metrics.get('throughput', 0),
            baseline_metrics.get('throughput', 0)
        )
        if throughput_drop > self.thresholds['throughput_drop']['critical']:
            issues.append(f"吞吐量下降过多：{throughput_drop:.2%}")
            score -= 30
        elif throughput_drop > self.thresholds['throughput_drop']['warning']:
            issues.append(f"吞吐量下降警告：{throughput_drop:.2%}")
            score -= 15
        
        # 4. 业务指标检查
        business_drop = self._calculate_drop(
            canary_metrics.get('conversion_rate', 0),
            baseline_metrics.get('conversion_rate', 0)
        )
        if business_drop > self.thresholds['business_metric_drop']['critical']:
            issues.append(f"业务指标下降过多：{business_drop:.2%}")
            score -= 40
        elif business_drop > self.thresholds['business_metric_drop']['warning']:
            issues.append(f"业务指标下降警告：{business_drop:.2%}")
            score -= 20
        
        # 确定状态与建议
        if score >= 80:
            status = 'healthy'
            recommendation = 'proceed'
        elif score >= 60:
            status = 'warning'
            recommendation = 'hold'
        else:
            status = 'critical'
            recommendation = 'rollback'
        
        return {
            'status': status,
            'score': max(0, score),
            'issues': issues,
            'recommendation': recommendation
        }
    
    def _calculate_drop(self, current: float, baseline: float) -> float:
        """计算下降比例"""
        if baseline == 0:
            return 0
        return (baseline - current) / baseline


# 使用示例
decision_engine = CanaryDecisionEngine()

canary_metrics = {
    'error_rate': 0.02,
    'p95_latency': 250,
    'throughput': 950,
    'conversion_rate': 0.048
}

baseline_metrics = {
    'error_rate': 0.005,
    'p95_latency': 180,
    'throughput': 1000,
    'conversion_rate': 0.050
}

result = decision_engine.evaluate_canary_health(
    canary_metrics, 
    baseline_metrics
)

print(f"健康度评分：{result['score']}")
print(f"状态：{result['status']}")
print(f"建议：{result['recommendation']}")
print(f"问题列表：{result['issues']}")

# 输出：
# 健康度评分：65
# 状态：warning
# 建议：hold
# 问题列表：['错误率警告：2.00%', 'P95 延迟警告：250ms']
                        

class RollbackEngine:
    """自动化回滚引擎"""
    
    def __init__(self):
        self.state = 'idle'
        self.rollback_history = []
    
    async def execute_rollback(
        self,
        trigger_reason: str,
        target_version: str = None,
        strategy: str = 'immediate'
    ) -> Dict:
        """
        执行回滚操作
        
        Args:
            trigger_reason: 触发原因
            target_version: 目标版本（默认上一个稳定版本）
            strategy: 回滚策略 ('immediate' | 'gradual')
        
        Returns:
            Dict: 回滚结果
        """
        self.state = 'rolling_back'
        start_time = time.time()
        
        try:
            # 1. 确定回滚目标版本
            if not target_version:
                target_version = await self._get_last_stable_version()
            
            # 2. 通知相关系统
            await self._notify_stakeholders(trigger_reason, target_version)
            
            # 3. 停止新版本流量
            await self._cut_traffic('new_version')
            
            # 4. 执行回滚策略
            if strategy == 'immediate':
                result = await self._immediate_rollback(target_version)
            else:  # gradual
                result = await self._gradual_rollback(target_version)
            
            # 5. 验证回滚成功
            success = await self._verify_rollback(target_version)
            
            # 6. 记录回滚历史
            rollback_record = {
                'timestamp': datetime.now(),
                'trigger_reason': trigger_reason,
                'from_version': result['from_version'],
                'to_version': target_version,
                'strategy': strategy,
                'duration': time.time() - start_time,
                'success': success
            }
            self.rollback_history.append(rollback_record)
            
            # 7. 发送回滚完成通知
            await self._notify_completion(rollback_record)
            
            self.state = 'idle'
            
            return {
                'success': success,
                'duration': rollback_record['duration'],
                'target_version': target_version
            }
            
        except Exception as e:
            self.state = 'failed'
            await self._notify_failure(str(e))
            raise
    
    async def _immediate_rollback(self, target_version: str) -> Dict:
        """立即回滚（秒级完成）"""
        
        # 1. 切换负载均衡配置
        await self._update_load_balancer(
            target_version,
            weight=100
        )
        
        # 2. 更新服务注册中心
        await self._update_service_registry(target_version)
        
        # 3. 等待配置生效
        await asyncio.sleep(2)
        
        # 4. 验证新版本无流量
        traffic = await self._check_traffic('new_version')
        assert traffic == 0, "新版本仍有流量"
        
        return {
            'from_version': 'new_version',
            'to_version': target_version,
            'method': 'immediate'
        }
    
    async def _gradual_rollback(self, target_version: str) -> Dict:
        """渐进式回滚（降低冲击）"""
        
        stages = [75, 50, 25, 0]  # 新版本流量比例
        
        for percentage in stages:
            # 更新流量比例
            await self._update_traffic_split(
                old_version=target_version,
                new_version='new_version',
                new_percentage=percentage
            )
            
            # 等待稳定
            await asyncio.sleep(30)
            
            # 检查指标
            metrics = await self._get_current_metrics()
            if metrics['error_rate'] > 0.01:
                # 指标异常，加速回滚
                await self._immediate_rollback(target_version)
                break
        
        return {
            'from_version': 'new_version',
            'to_version': target_version,
            'method': 'gradual'
        }
    
    async def _verify_rollback(self, target_version: str) -> bool:
        """验证回滚成功"""
        
        # 1. 检查版本一致性
        deployed_version = await self._get_deployed_version()
        if deployed_version != target_version:
            return False
        
        # 2. 检查健康指标
        metrics = await self._get_current_metrics()
        if metrics['error_rate'] > 0.01:
            return False
        if metrics['p95_latency'] > 500:
            return False
        
        # 3. 运行健康检查
        health_check = await self._run_health_checks()
        if not health_check['passed']:
            return False
        
        return True


# 使用示例
rollback_engine = RollbackEngine()

# 自动触发回滚
async def auto_rollback_trigger():
    # 监控指标
    metrics = await get_current_metrics()
    
    # 检查回滚条件
    if metrics['error_rate'] > 0.10:  # 错误率>10%
        await rollback_engine.execute_rollback(
            trigger_reason=f"错误率过高：{metrics['error_rate']:.2%}",
            strategy='immediate'
        )
    
    elif metrics['p99_latency'] > 5000:  # P99 延迟>5 秒
        await rollback_engine.execute_rollback(
            trigger_reason=f"P99 延迟过高：{metrics['p99_latency']}ms",
            strategy='immediate'
        )

# 手动触发回滚
async def manual_rollback(version: str, reason: str):
    await rollback_engine.execute_rollback(
        trigger_reason=reason,
        target_version=version,
        strategy='gradual'
    )
                        

class BlueGreenDeployment:
    """蓝绿部署管理器"""
    
    def __init__(self):
        self.environments = {
            'blue': {'version': None, 'status': 'inactive', 'instances': []},
            'green': {'version': None, 'status': 'inactive', 'instances': []}
        }
        self.active_env = None
    
    async def deploy_new_version(self, version: str, image: str) -> str:
        """
        部署新版本到空闲环境
        
        Returns:
            str: 部署的环境名称 ('blue' 或 'green')
        """
        # 1. 确定空闲环境
        target_env = 'blue' if self.active_env == 'green' else 'green'
        
        # 2. 部署新版本
        print(f"部署版本 {version} 到 {target_env} 环境")
        instances = await self._create_instances(
            environment=target_env,
            version=version,
            image=image,
            count=3
        )
        
        # 3. 健康检查
        health_passed = await self._health_check(instances)
        if not health_passed:
            await self._cleanup_instances(instances)
            raise Exception("健康检查失败")
        
        # 4. 更新环境状态
        self.environments[target_env].update({
            'version': version,
            'status': 'ready',
            'instances': instances
        })
        
        return target_env
    
    async def switch_traffic(self, target_env: str) -> None:
        """
        切换流量到目标环境
        
        Args:
            target_env: 目标环境 ('blue' 或 'green')
        """
        if target_env not in ['blue', 'green']:
            raise ValueError("Invalid environment")
        
        old_env = self.active_env
        
        # 1. 更新负载均衡
        await self._update_load_balancer(
            target_env=target_env,
            weight=100
        )
        
        # 2. 更新 DNS（如果需要）
        await self._update_dns(target_env)
        
        # 3. 更新活动环境
        self.active_env = target_env
        self.environments[target_env]['status'] = 'active'
        
        if old_env:
            self.environments[old_env]['status'] = 'inactive'
        
        print(f"流量已从 {old_env} 切换到 {target_env}")
    
    async def rollback(self) -> None:
        """快速回滚到上一个环境"""
        if not self.active_env:
            raise Exception("No active deployment")
        
        # 确定上一个环境
        previous_env = 'blue' if self.active_env == 'green' else 'green'
        
        if self.environments[previous_env]['status'] != 'ready':
            raise Exception("Previous environment not available")
        
        # 切换流量
        await self.switch_traffic(previous_env)
        
        print(f"回滚完成，当前活动环境：{previous_env}")
    
    async def cleanup_inactive(self) -> None:
        """清理非活动环境释放资源"""
        for env_name, env_data in self.environments.items():
            if env_name != self.active_env and env_data['status'] == 'inactive':
                await self._cleanup_instances(env_data['instances'])
                env_data['instances'] = []
                env_data['version'] = None
                print(f"已清理 {env_name} 环境")


# 使用示例
bg_deployment = BlueGreenDeployment()

# 初始部署
async def initial_deployment():
    # 部署到 blue 环境
    await bg_deployment.deploy_new_version(
        version='v1.0.0',
        image='myapp:v1.0.0'
    )
    await bg_deployment.switch_traffic('blue')

# 新版本发布
async def release_new_version():
    # 部署 v2.0.0 到 green 环境
    target_env = await bg_deployment.deploy_new_version(
        version='v2.0.0',
        image='myapp:v2.0.0'
    )
    
    # 验证 green 环境
    metrics = await get_environment_metrics(target_env)
    if metrics['error_rate'] < 0.01:
        # 切换流量到 green
        await bg_deployment.switch_traffic(target_env)
        # 清理 blue 环境
        await bg_deployment.cleanup_inactive()
    else:
        # 发现问题，清理 green 环境
        await bg_deployment.cleanup_inactive()
        raise Exception("新版本有问题，发布失败")

# 快速回滚
async def emergency_rollback():
    await bg_deployment.rollback()
    print("已回滚到上一个稳定版本")
                        

# prometheus.yml 配置文件
global:
  scrape_interval: 15s      # 采集间隔
  evaluation_interval: 15s  # 规则评估间隔

scrape_configs:
  # 采集 Agent 服务指标
  - job_name: 'agent-service'
    static_configs:
      - targets: ['agent-service:8080']
    metrics_path: '/metrics'
    
  # 采集金丝雀版本指标
  - job_name: 'canary-version'
    static_configs:
      - targets: ['agent-canary:8080']
    metrics_path: '/metrics'
    relabel_configs:
      - source_labels: [__address__]
        target_label: version
        replacement: 'canary'
  
  # 采集稳定版本指标
  - job_name: 'stable-version'
    static_configs:
      - targets: ['agent-stable:8080']
    metrics_path: '/metrics'
    relabel_configs:
      - source_labels: [__address__]
        target_label: version
        replacement: 'stable'

# 告警规则
rule_files:
  - 'alert_rules.yml'

# Alertmanager 配置
alerting:
  alertmanagers:
    - static_configs:
        - targets: ['alertmanager:9093']
                        

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
import random

# 定义指标
REQUEST_COUNT = Counter(
    'agent_requests_total',
    'Agent 总请求数',
    ['version', 'endpoint', 'status']
)

REQUEST_LATENCY = Histogram(
    'agent_request_latency_seconds',
    'Agent 请求延迟',
    ['version', 'endpoint'],
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
)

ACTIVE_CONNECTIONS = Gauge(
    'agent_active_connections',
    '活跃连接数',
    ['version']
)

ERROR_RATE = Gauge(
    'agent_error_rate',
    '错误率',
    ['version']
)

CONVERSION_RATE = Gauge(
    'agent_conversion_rate',
    '业务转化率',
    ['version']
)

class AgentMetricsCollector:
    """Agent 指标采集器"""
    
    def __init__(self, version: str):
        self.version = version
        self.error_count = 0
        self.total_count = 0
        self.success_count = 0
    
    def record_request(
        self, 
        endpoint: str, 
        latency: float, 
        status: str
    ):
        """记录请求指标"""
        self.total_count += 1
        
        # 记录请求计数
        REQUEST_COUNT.labels(
            version=self.version,
            endpoint=endpoint,
            status=status
        ).inc()
        
        # 记录请求延迟
        REQUEST_LATENCY.labels(
            version=self.version,
            endpoint=endpoint
        ).observe(latency)
        
        # 更新错误率
        if status.startswith('5'):
            self.error_count += 1
        
        # 更新转化率（假设成功完成业务逻辑算转化）
        if status == '200':
            self.success_count += 1
        
        # 计算并更新错误率
        if self.total_count > 0:
            error_rate = self.error_count / self.total_count
            ERROR_RATE.labels(version=self.version).set(error_rate)
        
        # 计算并更新转化率
        if self.total_count > 0:
            conversion_rate = self.success_count / self.total_count
            CONVERSION_RATE.labels(version=self.version).set(conversion_rate)
    
    def update_active_connections(self, count: int):
        """更新活跃连接数"""
        ACTIVE_CONNECTIONS.labels(version=self.version).set(count)


# 使用示例
metrics_collector = AgentMetricsCollector(version='v2.0.0')

# 启动指标服务器
start_http_server(8000)

# 模拟请求处理
def handle_request(endpoint: str):
    start_time = time.time()
    
    try:
        # 模拟业务逻辑
        time.sleep(random.uniform(0.05, 0.3))
        
        # 模拟随机错误
        if random.random() < 0.02:
            status = '500'
        else:
            status = '200'
        
    except Exception as e:
        status = '500'
    
    latency = time.time() - start_time
    
    # 记录指标
    metrics_collector.record_request(endpoint, latency, status)
    
    return status

# 持续运行
while True:
    handle_request('/api/chat')
    time.sleep(0.1)
                        

groups:
  - name: agent_alerts
    rules:
      # 错误率告警
      - alert: HighErrorRate
        expr: rate(agent_requests_total{status=~"5.."}[5m]) 
              / rate(agent_requests_total[5m]) > 0.05
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "错误率过高 ({{ $labels.version }})"
          description: "版本 {{ $labels.version }} 的错误率为 {{ $value | humanizePercentage }}"
      
      # P95 延迟告警
      - alert: HighP95Latency
        expr: histogram_quantile(0.95, 
               rate(agent_request_latency_seconds_bucket[5m])) > 1
        for: 3m
        labels:
          severity: warning
        annotations:
          summary: "P95 延迟过高 ({{ $labels.version }})"
          description: "版本 {{ $labels.version }} 的 P95 延迟为 {{ $value | humanizeDuration }}"
      
      # 金丝雀版本异常
      - alert: CanaryVersionAnomaly
        expr: |
          (
            rate(agent_requests_total{version="canary",status=~"5.."}[5m]) 
            / rate(agent_requests_total{version="canary"}[5m])
          ) > 3 * (
            rate(agent_requests_total{version="stable",status=~"5.."}[5m]) 
            / rate(agent_requests_total{version="stable"}[5m])
          )
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "金丝雀版本异常"
          description: "金丝雀版本错误率是稳定版本的 3 倍以上"
      
      # 转化率下降告警
      - alert: ConversionRateDrop
        expr: |
          (
            agent_conversion_rate{version="stable"} 
            - agent_conversion_rate{version="canary"}
          ) / agent_conversion_rate{version="stable"} > 0.15
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "转化率下降"
          description: "金丝雀版本转化率比稳定版本下降超过 15%"
      
      # 活跃连接数过高
      - alert: HighActiveConnections
        expr: agent_active_connections > 1000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "活跃连接数过高"
          description: "版本 {{ $labels.version }} 的活跃连接数为 {{ $value }}"