全链路灰度与混沌工程实战

apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: order-service-vs
  namespace: production
spec:
  hosts:
    - order-service
  http:
    # 灰度路由：带 x-canary-tag=gray-v2.3.1 Header 的请求
    - name: "canary-route"
      match:
        - headers:
            x-canary-tag:
              exact: "gray-v2.3.1"
      route:
        - destination:
            host: order-service
            subset: v2.3.1-canary
          weight: 100
    # Header 透传：必须将灰度标识传递到下游所有服务
    - name: "header-propagation"
      match:
        - headers:
            x-canary-tag:
              exact: "gray-v2.3.1"
      route:
        - destination:
            host: order-service
            subset: stable
      headers:
        request:
          set:
            x-canary-tag: "gray-v2.3.1"
    # 权重路由：10% 流量切到灰度版本（用于 A/B 测试）
    - name: "weight-route"
      match:
        - headers:
            x-canary-tag:
              exact: "abtest"
      route:
        - destination:
            host: order-service
            subset: stable
          weight: 90
        - destination:
            host: order-service
            subset: canary-latest
          weight: 10
    # 主流量路由：无灰度标识走稳定版本
    - name: "stable-route"
      route:
        - destination:
            host: order-service
            subset: stable
          weight: 100
---
# DestinationRule 定义版本子集
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: order-service-dr
  namespace: production
spec:
  host: order-service
  trafficPolicy:
    tls:
      mode: ISTIO_MUTUAL  # mTLS 强制开启
  subsets:
    - name: stable
      labels:
        version: stable
    - name: canary-latest
      labels:
        version: canary
    - name: v2.3.1-canary
      labels:
        version: v2.3.1-canary
        canary-tag: gray-v2.3.1

apiVersion: networking.istio.io/v1alpha3
kind: EnvoyFilter
metadata:
  name: canary-header-propagation
  namespace: istio-system
spec:
  workloadSelector:
    labels:
      # 作用于所有 Sidecar
      istio: sidecar
  configPatches:
    - applyTo: HTTP_FILTER
      match:
        context: SIDECAR_OUTBOUND
        listener:
          filterChain:
            filter:
              name: envoy.filters.network.http_connection_manager
      patch:
        operation: INSERT_BEFORE
        value:
          name: envoy.filters.http.lua
          typed_config:
            "@type": type.googleapis.com/envoy.extensions.filters.http.lua.v3.Lua
            inlineCode: |
              function envoy_on_request(request_handle)
                -- 读取入口灰度标识
                local canary_tag = request_handle:headers():get("x-canary-tag")
                if canary_tag and canary_tag ~= "" then
                  -- 透传给所有下游服务
                  request_handle:headers():add("x-canary-tag", canary_tag)
                  request_handle:headers():add("x-request-canary", "true")
                end
              end
              function envoy_on_response(response_handle)
                -- 在响应头中标记当前版本
                response_handle:headers():add("x-served-by-canary", "true")
              end

apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: payment-service-shadow
spec:
  hosts:
    - payment-service
  http:
    - name: "shadow-mirror"
      route:
        # 90% 流量走稳定版本
        - destination:
            host: payment-service
            subset: stable
          weight: 90
        # 10% 流量走稳定版本，但同时镜像 10% 到灰度版本
        - destination:
            host: payment-service
            subset: stable
          weight: 10
      mirrors:
        - destination:
            host: payment-service
            subset: canary-latest
          # 镜像比例 100%，但不影响主流量
          percentage:
            value: 100
      retries:
        destination:
          host: payment-service
          subset: stable
        attempts: 2
        perTryTimeout: 5s

@Configuration
public class CircuitBreakerConfig {

    // 为每个下游服务创建独立的熔断器实例
    @Bean
    public CircuitBreakerRegistry circuitBreakerRegistry() {
        CircuitBreakerConfig defaultConfig = CircuitBreakerConfig.custom()
            // 滑动窗口大小：最近 100 次调用
            .slidingWindowSize(100)
            // 最小调用数：至少 50 次调用才计算失败率
            .minimumNumberOfCalls(50)
            // 失败率阈值：超过 50% 失败率则熔断
            .failureRateThreshold(50)
            // 慢调用阈值：超过 2 秒的调用视为慢调用
            .slowCallDurationThreshold(Duration.ofSeconds(2))
            // 慢调用失败率阈值：慢调用超过 80% 则熔断
            .slowCallRateThreshold(80)
            // 熔断持续时间：熔断后 30 秒进入半开状态
            .waitDurationInOpenState(Duration.ofSeconds(30))
            // 半开状态允许通过的请求数
            .permittedNumberOfCallsInHalfOpenState(10)
            // 自动从打开切换到半开
            .automaticTransitionFromOpenToHalfOpenEnabled(true)
            .build();

        return CircuitBreakerRegistry.of(defaultConfig);
    }

    // 高频服务使用更严格的熔断策略
    @Bean("strictCircuitBreaker")
    public CircuitBreakerRegistry strictCircuitBreakerRegistry() {
        CircuitBreakerConfig strictConfig = CircuitBreakerConfig.custom()
            .slidingWindowSize(50)
            .minimumNumberOfCalls(20)
            .failureRateThreshold(30)   // 更敏感，30% 即熔断
            .slowCallDurationThreshold(Duration.ofSeconds(1))
            .slowCallRateThreshold(50)
            .waitDurationInOpenState(Duration.ofSeconds(15))
            .permittedNumberCallsInHalfOpenState(5)
            .build();
        return CircuitBreakerRegistry.of(strictConfig);
    }
}

// 在业务代码中使用熔断器保护外部服务调用
@Service
public class OrderService {

    private final CircuitBreaker inventoryCircuitBreaker;
    private final InventoryClient inventoryClient;
    private final RedisTemplate<String, String> redisTemplate;

    public OrderService(CircuitBreakerRegistry registry,
                        InventoryClient inventoryClient,
                        RedisTemplate<String, String> redisTemplate) {
        this.inventoryCircuitBreaker = registry.circuitBreaker("inventory-service");
        this.inventoryClient = inventoryClient;
        this.redisTemplate = redisTemplate;
    }

    public Order createOrder(OrderRequest request) {
        Supplier<InventoryResult> decoratedSupplier = CircuitBreaker.decorateSupplier(
            inventoryCircuitBreaker,
            () -> inventoryClient.reserveStock(request.getSkuId(), request.getQty())
        );

        Try<InventoryResult> result = Try.ofSupplier(decoratedSupplier)
            // 降级逻辑：当熔断触发时，使用 Redis 缓存的库存快照
            .recover(throwable -> {
                if (throwable instanceof CallNotPermittedException) {
                    log.warn("Inventory service circuit OPEN, using cached data");
                    return getInventoryFromCache(request.getSkuId());
                }
                if (throwable instanceof TimeoutException) {
                    log.warn("Inventory service timeout, fallback to cached data");
                    return getInventoryFromCache(request.getSkuId());
                }
                throw new RuntimeException(throwable);
            });

        // 执行主业务逻辑
        return doCreateOrder(request, result.get());
    }

    // 从 Redis 缓存获取库存快照（降级策略）
    private InventoryResult getInventoryFromCache(String skuId) {
        String cached = redisTemplate.opsForValue().get("inventory:snapshot:" + skuId);
        if (cached != null) {
            return JSON.parseObject(cached, InventoryResult.class);
        }
        // 最坏情况：返回无限库存，依赖下游超时的最终一致性
        return InventoryResult.unlimited();
    }
}

@Service
public class FallbackChain {

    private final RedisTemplate<String, Object> redis;
    private final LocalHotDataService localHotData;
    private final DefaultResultFactory defaultResultFactory;

    public <T> T executeFallbackChain(String serviceName, String cacheKey,
                                        Supplier<T> primaryCall, Class<T> resultType) {
        try {
            // L1: 尝试主调用
            T result = primaryCall.get();
            if (result != null) {
                // 异步回填 Redis 缓存（不阻塞主流程）
                asyncCacheWrite(cacheKey, result);
                return result;
            }
        } catch (Exception e) {
            log.warn("Primary call failed for {}, fallback to L1 cache", serviceName, e);
        }

        // L2: Redis 缓存
        T cached = redis.opsForValue().get(cacheKey);
        if (cached != null) {
            log.info("L2 fallback hit for {}", serviceName);
            return cached;
        }

        // L3: 本地热点数据兜底
        T localData = localHotData.get(serviceName, resultType);
        if (localData != null) {
            log.warn("L3 fallback hit for {} (local data)", serviceName);
            return localData;
        }

        // L4: 返回默认值工厂的兜底数据
        log.error("All fallback levels exhausted for {}", serviceName);
        return defaultResultFactory.create(serviceName, resultType);
    }

    @Async
    private void asyncCacheWrite(String cacheKey, Object value) {
        try {
            redis.opsForValue().set(cacheKey, value, 5, TimeUnit.MINUTES);
        } catch (Exception e) {
            log.warn("Failed to write cache for {}", cacheKey, e);
        }
    }
}

3.4 Flink 实时特征更新（灰度决策数据源）

灰度发布的流量分配策略不仅要基于版本号，还需要结合实时业务指标（如用户活跃度、订单金额）动态调整。我们使用 Flink 实时计算用户价值分，并将其作为灰度流量分配的辅助信号：高价值用户默认走稳定版本，新用户可以尝试灰度版本收集行为数据。

public class UserValueScoreJob {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(30_000); // 30s checkpoint
        env.setStateBackend(new RocksDBStateBackend("hdfs:///flink/checkpoints"));
        env.getConfig().setAutoWatermarkInterval(1000);

        // 数据源：Kafka 用户行为事件
        DataStream<UserEvent> eventStream = env
            .addSource(new FlinkKafkaConsumer<>(
                "user-behavior-events",
                new UserEventDeserializationSchema(),
                kafkaProps
            ))
            .keyBy(UserEvent::getUserId)
            .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)))
            .process(new UserValueScoreCalculator());

        // 输出到 Redis，供灰度决策服务查询
        eventStream.addSink(new RedisSink<>(
            redisConfig,
            new UserScoreRedisMapper()
        ));

        env.execute("UserValueScoreJob");
    }

    // 用户价值评分计算
    public static class UserValueScoreCalculator
            extends KeyedProcessFunction<String, UserEvent, UserScore> {

        private ValueState<UserScoreAccumulator> accumulator;

        @Override
        public void open(Configuration params) {
            accumulator = getRuntimeContext().getState(
                new ValueStateDescriptor<>("user-score-acc",
                    UserScoreAccumulator.class)
            );
        }

        @Override
        public void processElement(UserEvent event, Context ctx, Collector<UserScore> out)
                throws Exception {
            UserScoreAccumulator acc = accumulator.value();
            if (acc == null) {
                acc = new UserScoreAccumulator();
            }

            // 累计各项指标
            switch (event.getEventType()) {
                case "order":
                    acc.orderCount++;
                    acc.totalOrderAmount += event.getAmount();
                    acc.lastOrderTime = event.getTimestamp();
                    break;
                case "browse":
                    acc.browseCount++;
                    break;
                case "cart":
                    acc.cartCount++;
                    break;
            }
            acc.lastActiveTime = event.getTimestamp();

            // 计算综合评分（0-100）
            double score = calculateScore(acc);
            accumulator.update(acc);

            out.collect(new UserScore(
                event.getUserId(),
                score,
                acc.tier,  // GOLD/SILVER/BRONZE/NEW
                ctx.timestamp()
            ));
        }

        private double calculateScore(UserScoreAccumulator acc) {
            double orderScore = Math.min(acc.totalOrderAmount / 10000.0, 40); // 最高40分
            double frequencyScore = Math.min(acc.orderCount / 10.0, 30);     // 最高30分
            double recencyScore = calculateRecencyScore(acc.lastOrderTime); // 最高30分
            return orderScore + frequencyScore + recencyScore;
        }
    }
}

apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
  name: order-service-delay-chaos
  namespace: chaos-testing
spec:
  action: delay
  mode: one
  duration: "10m"       # 持续 10 分钟
  delay:
    latency: "500ms"     # 注入 500ms 固定延迟
    correlation: "25"    # 延迟抖动系数（0-100）
    jitter: "200ms"      # 随机抖动范围
  selector:
    namespaces:
      - production
    labelSelectors:
      app: inventory-service  # 目标服务：库存服务
  # 排除 ChaosMesh Sidecar，避免自身被干扰
  chaosMetadata:
    namespace: chaos-testing
---
# 实验告警规则：当订单失败率超过 1% 时自动终止实验
apiVersion: chaos-mesh.org/v1alpha1
kind: Schedule
metadata:
  name: weekly-network-chaos
  namespace: chaos-testing
spec:
  schedule: "0 2 * * 1"  # 每周一凌晨 2 点执行
  type: Schedule
  workflow:
    templateName: network-chaos-workflow
  concurrencyPolicy: Forbid
---
# 混沌实验 Workflow：注入 → 观察 → 回滚
apiVersion: chaos-mesh.org/v1alpha1
kind: Workflow
metadata:
  name: network-chaos-workflow
  namespace: chaos-testing
spec:
  entry: network-chaos-sequential
  templates:
    - name: network-chaos-sequential
      type: Sequential
      children:
        - name: inject-delay
          template: inject-500ms-delay
        - name: observe
          template: observe-5min
        - name: rollback
          template: cleanup-chaos
    - name: inject-500ms-delay
      type: NetworkChaos
      # ... 注入配置
    - name: observe-5min
      type: StubHTTP
      duration: 5m
      # 持续观察订单成功率
    - name: cleanup-chaos
      type: Noop  # 等待 duration 结束后自动清理

# CPU 压力实验：模拟高负载
apiVersion: chaos-mesh.org/v1alpha1
kind: StressChaos
metadata:
  name: order-service-cpu-stress
  namespace: chaos-testing
spec:
  mode: one
  duration: "5m"
  stressors:
    cpu:
      # 在 2 个 CPU 核心上施加 80% 压力
      workers: 2
      load: 80
  selector:
    namespaces:
      - production
    labelSelectors:
      app: order-service
      version: stable

# Pod Kill 实验：模拟 K8s Pod 驱逐
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
  name: payment-pod-kill-chaos
  namespace: chaos-testing
spec:
  action: pod-kill
  mode: fixed
  value: "1"           # 杀 1 个 Pod（保持最小可用数）
  duration: "3m"
  selector:
    namespaces:
      - production
    labelSelectors:
      app: payment-service
  # 与 PodDisruptionBudget 配合，确保最小可用数
  pdb:
    maxUnavailable: 1

apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: order-chain-gray
  namespace: production
spec:
  hosts:
    - order-gateway
  http:
    # ============ 策略一：Header 精确匹配灰度 ============
    - name: "header-canary"
      match:
        - headers:
            x-canary-tag:
              exact: "gray-v2.4.0"
          source_labels:
            version: stable
      route:
        - destination:
            host: order-service
            subset: v2.4.0-canary
          weight: 100
      retries:
        attempts: 3
        perTryTimeout: 3s
        retryOn: connect-failure,refused-stream,unavailable,cancelled,retriable-status-codes
      timeout: 10s
      # ============ 策略二：10% 流量权重灰度（A/B 测试）============
    - name: "weight-abtest"
      match:
        - headers:
            x-canary-mode:
              exact: "abtest"
      route:
        - destination:
            host: order-service
            subset: stable
          weight: 90
        - destination:
            host: order-service
            subset: canary-latest
          weight: 10
      # ============ 策略三：用户分桶灰度（按用户ID哈希）============
    - name: "user-hash-canary"
      match:
        - headers:
            x-user-hash:
              regex: "^[0-2].*"   # 前10%的用户ID哈希走灰度
      route:
        - destination:
            host: order-service
            subset: canary-latest
          weight: 100
      # ============ 默认稳定版本 ============
    - name: "stable-default"
      route:
        - destination:
            host: order-service
            subset: stable
          weight: 100
---
# 熔断配置（全局）
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: order-service-breaker
spec:
  host: order-service
  trafficPolicy:
    outlierDetection:
      # 连续 5 次失败则驱逐该实例
      consecutive5xxErrors: 5
      # 每 30 秒检测一次
      interval: 30s
      # 驱逐后 60 秒恢复
      baseEjectionTime: 60s
      # 最大驱逐比例 50%
      maxEjectionPercent: 50
    connectionPool:
      http:
        h2UpgradePolicy: UPGRADE
        http1MaxPendingRequests: 100
        http2MaxRequests: 1000
        maxRequestsPerConnection: 100
    loadBalancer:
      simple: LEAST_REQUEST
      localityLbSetting:
        enabled: true