# Nginx示例配置片段
limit_conn_zone $binary_remote_addr zone=ws_conn:10m;
limit_req_zone $binary_remote_addr zone=ws_handshake:10m rate=15r/s;

server {
    location /signaling {
        limit_conn ws_conn 80;
        limit_req zone=ws_handshake burst=30 nodelay;
        proxy_read_timeout 300s;
        proxy_send_timeout 300s;
    }
}

CLOSED (正常) --触发条件达标--> OPEN (熔断)
OPEN --冷却期(建议30-60s)--> HALF_OPEN (半开)
HALF_OPEN --探测请求成功率达标--> CLOSED
HALF_OPEN --探测失败--> OPEN (重置冷却期)

// token_bucket.go - 生产级无锁令牌桶
type LockFreeTokenBucket struct {
    capacity     int64           // 桶容量
    ratePerSec   int64           // 每秒填充速率
    tokens       int64           // 当前令牌数 (原子操作)
    lastRefillNs int64           // 上次填充时间纳秒戳 (原子操作)
}

// Take 尝试获取n个令牌，非阻塞，返回是否成功
func (b *LockFreeTokenBucket) Take(n int64) bool {
    for {
        now := time.Now().UnixNano()
        last := atomic.LoadInt64(&b.lastRefillNs)
        tokens := atomic.LoadInt64(&b.tokens)

        // 计算应补充令牌数
        elapsed := now - last
        refill := (elapsed * b.ratePerSec) / 1e9
        newTokens := tokens + refill
        if newTokens > b.capacity {
            newTokens = b.capacity
        }

        if newTokens < n {
            return false // 令牌不足
        }

        // CAS尝试更新状态
        if atomic.CompareAndSwapInt64(&b.lastRefillNs, last, now) &&
           atomic.CompareAndSwapInt64(&b.tokens, tokens, newTokens-n) {
            return true
        }
        // CAS失败自旋重试（极低概率）
        runtime.Gosched()
    }
}

-- sliding_window.lua
-- KEYS[1]: Redis Key (格式: ratelimit:{biz}:{user_id}:{window_sec})
-- ARGV[1]: 当前时间戳毫秒
-- ARGV[2]: 窗口大小毫秒
-- ARGV[3]: 限流阈值
-- ARGV[4]: 请求权重 (通常为1)

local key = KEYS[1]
local now = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])
local weight = tonumber(ARGV[4])

-- 1. 移除窗口外的旧数据
redis.call('ZREMRANGEBYSCORE', key, 0, now - window)

-- 2. 计算当前窗口内总权重
local current = redis.call('ZRANGE', key, 0, -1, 'WITHSCORES')
local total_weight = 0
for i = 2, #current, 2 do
    total_weight = total_weight + tonumber(current[i])
end

-- 3. 判断是否超限
if total_weight + weight > limit then
    -- 返回剩余TTL供客户端Retry-After参考
    local ttl = redis.call('PTTL', key)
    if ttl < 0 then ttl = window end
    return {0, total_weight, ttl} -- 0=拒绝
end

-- 4. 写入新请求 (Score=时间戳, Member=唯一请求ID防重复计数)
local request_id = now .. "." .. math.random(1000000)
redis.call('ZADD', key, now, request_id)
redis.call('PEXPIRE', key, window)

return {1, total_weight + weight, window} -- 1=通过

func (l *DistributedLimiter) Allow(ctx context.Context, key string, weight int) (allowed bool, current int, retryAfterMs int, err error) {
    res, err := l.redis.EvalSha(ctx, l.sha, []string{key}, 
        time.Now().UnixMilli(), l.windowMs, l.limit, weight).Slice()
    if err != nil { return false, 0, 0, err }
    allowed = res[0].(int64) == 1
    current = int(res[1].(int64))
    retryAfterMs = int(res[2].(int64))
    return
}

// breaker.go - 核心状态流转逻辑片段
func (cb *CircuitBreaker) execute(req func() (interface{}, error)) (interface{}, error) {
    state := atomic.LoadInt32(&cb.state)
    
    switch State(state) {
    case StateOpen:
        // 冷却期未到，直接拒绝
        if time.Since(cb.openedAt) < cb.config.Timeout {
            return nil, ErrCircuitOpen
        }
        // 进入半开，仅允许单个探测请求通过（互斥锁保护）
        if !atomic.CompareAndSwapInt32(&cb.state, int32(StateOpen), int32(StateHalfOpen)) {
            return nil, ErrCircuitOpen // 抢占失败，当作熔断处理
        }
        cb.halfOpenSuccesses.Store(0) // 重置半开成功计数
        fallthrough // 继续执行探测请求

    case StateHalfOpen:
        // 半开状态：限流器仅放行极小比例流量 (如 1%)
        if !cb.halfOpenLimiter.Allow() { 
            return nil, ErrCircuitHalfOpenBusy
        }
    }

    result, err := req()
    
    // 结果记录与状态流转 (简化版)
    cb.recordResult(err == nil)
    return result, err
}

func (cb *CircuitBreaker) recordResult(success bool) {
    // ... 统计成功/失败计数 ...
    // 闭合->打开：失败率阈值 + 最小请求数门槛
    // 半开->闭合：连续 N 次成功
    // 半开->打开：任意一次失败
}

// signaling-client.js - 标准重试策略
class SignalingRetryPolicy {
    constructor(config = {}) {
        this.baseDelay = config.baseDelay || 1000;      // 基础延迟 1s
        this.maxDelay = config.maxDelay || 30000;       // 最大延迟 30s
        this.maxRetries = config.maxRetries || 5;       // 最大重试次数
        this.jitterFactor = config.jitterFactor || 0.3; // 抖动因子 30%
    }

    calculateDelay(attempt, serverRetryAfter) {
        // 1. 优先遵守服务端 Retry-After (秒)
        if (serverRetryAfter && serverRetryAfter > 0) {
            return serverRetryAfter * 1000 + this.randomJitter(1000);
        }
        // 2. 指数退避: base * 2^attempt
        const exponentialDelay = this.baseDelay * Math.pow(2, attempt);
        // 3. 全抖动: random(0, delay) - 比等抖动更平滑
        const jitteredDelay = Math.random() * exponentialDelay;
        return Math.min(jitteredDelay, this.maxDelay);
    }

    randomJitter(range) { return Math.random() * range; }

    async executeWithRetry(requestFn, onRetry) {
        for (let attempt = 0; attempt <= this.maxRetries; attempt++) {
            try {
                const response = await requestFn();
                // 处理 503/429 等可重试状态码
                if (this.isRetriable(response)) {
                    const delay = this.calculateDelay(attempt, response.headers.get('Retry-After'));
                    if (onRetry) onRetry(attempt, delay, response);
                    await this.sleep(delay);
                    continue;
                }
                return response;
            } catch (err) {
                if (attempt === this.maxRetries || !this.isNetworkError(err)) throw err;
                const delay = this.calculateDelay(attempt);
                if (onRetry) onRetry(attempt, delay, err);
                await this.sleep(delay);
            }
        }
    }
}

// 客户端熔断状态机
enum ClientCircuitState { Healthy, Degraded, Offline }

class ClientCircuitBreaker {
    private state = ClientCircuitState.Healthy;
    private failureCount = 0;
    
    onSignalError(error: SignalingError) {
        if (error.code === 'SIGNALING_OVERLOAD' || error.code === 'SERVER_CIRCUIT_OPEN') {
            this.failureCount++;
            if (this.failureCount > 3 && this.state === ClientCircuitState.Healthy) {
                this.transitionToDegraded(); // 降级：关闭视频上行、降低码率、暂停屏幕共享
            }
        }
    }
    
    onSignalSuccess() {
        this.failureCount = 0;
        if (this.state === ClientCircuitState.Degraded) this.transitionToHealthy();
    }
    
    private transitionToDegraded() {
        this.state = ClientCircuitState.Degraded;
        this.eventBus.emit('media:downgrade', { reason: 'signaling_overload' });
        // UI 提示：网络拥塞，已自动降低画质保障通话连续性
    }
}

# configmap-signaling.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: signaling-ratelimit-config
data:
  # 单 Pod 允许的最大连接数 = 总目标连接数 / 当前副本数
  max_connections_per_pod: "10000" 
  # 单 Pod QPS 限流 = 总目标 QPS / 当前副本数 * 安全系数(0.8)
  qps_limit_per_pod: "500"

// k8s-controller/main.go - 简化版控制器逻辑
func (c *Controller) syncReplicasAndLimits() {
    deployment, _ := c.k8sClient.AppsV1().Deployments(ns).Get(context.TBG, "signaling-server", metav1.GetOptions{})
    replicas := *deployment.Spec.Replicas
    if replicas == 0 { return }

    // 从全局配置读取总容量目标
    totalConnTarget := c.globalConfig.TotalMaxConnections // 如 200,000
    totalQPSTarget := c.globalConfig.TotalTargetQPS       // 如 50,000

    // 计算单 Pod 配额 (预留 20% 缓冲)
    perPodConn := int64(float64(totalConnTarget) / float64(replicas) * 0.8)
    perPodQPS := int64(float64(totalQPSTarget) / float64(replicas) * 0.8)

    // 更新 ConfigMap (触发应用热加载)
    cm, _ := c.k8sClient.CoreV1().ConfigMaps(ns).Get(context.TBG, "signaling-ratelimit-config", metav1.GetOptions{})
    cm.Data["max_connections_per_pod"] = strconv.FormatInt(perPodConn, 10)
    cm.Data["qps_limit_per_pod"] = strconv.FormatInt(perPodQPS, 10)
    c.k8sClient.CoreV1().ConfigMaps(ns).Update(context.TBG, cm, metav1.UpdateOptions{})
}

// conn_manager.go - 慢速连接清理器
func (m *ConnManager) startSlowConnectionReaper() {
    ticker := time.NewTicker(30 * time.Second)
    for range ticker.C {
        m.conns.Range(func(key, val interface{}) bool {
            conn := val.(*SignalingConn)
            // 统计窗口内有效载荷字节数
            if conn.bytesReadInWindow < 100 && conn.messagesReadInWindow < 1 {
                log.Warn("Slow connection detected, closing", "remote", conn.RemoteAddr())
                conn.Close(ClosePolicyViolation, "slowloris_detected")
            }
            conn.resetWindowCounters()
            return true
        })
    }
}

// logger/sanitizer.go
var piiPatterns = []*regexp.Regexp{
    regexp.MustCompile(`(?i)"phone"s*:s*"(d{11})"`),
    regexp.MustCompile(`(?i)"id_card"s*:s*"([dX]{18})"`),
    regexp.MustCompile(`(?i)"real_name"s*:s*"([^"]+)"`),
}

func SanitizeLogMessage(msg string) string {
    for _, re := range piiPatterns {
        msg = re.ReplaceAllStringFunc(msg, func(match string) string {
            // 保留字段名，值替换为 ***
            return re.ReplaceAllString(match, `${1}"***"`)
        })
    }
    return msg
}

// 结构化日志字段白名单机制
type LogFields map[string]interface{}
func (l LogFields) Safe() LogFields {
    allowed := []string{"user_id_hash", "room_id", "action", "latency_ms", "error_code", "client_version", "region"}
    safe := make(LogFields)
    for _, k := range allowed {
        if v, ok := l[k]; ok { safe[k] = v }
    }
    return safe
}

# chaos-experiment-signaling-pod-kill.yaml
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
  name: signaling-pod-kill-random
spec:
  action: pod-failure
  mode: one
  selector:
    namespaces: ["production"]
    labelSelectors:
      "app": "signaling-server"
  duration: "30s"
  scheduler:
    cron: "@every 1h" # 每小时随机杀一个 Pod
---
# chaos-experiment-network-partition.yaml
apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
  name: signaling-network-partition
spec:
  action: partition
  mode: all
  selector:
    namespaces: ["production"]
    labelSelectors:
      "app": "signaling-server"
  direction: both
  target:
    selector:
      namespaces: ["production"]
      labelSelectors:
        "app": "media-server" # 模拟信令与媒体服务器网络分区
  duration: "60s"