Kitex熔断

package circuitbreaker

import (
	"context"
	"sync/atomic"
	"time"

	"github.com/cloudwego/kitex/pkg/endpoint"
	"github.com/cloudwego/kitex/pkg/klog"
)

// State 熔断器状态
type State int

const (
	StateClosed   State = 0 // 闭合：正常
	StateOpen     State = 1 // 打开：熔断
	StateHalfOpen State = 2 // 半开：试探
)

// Config 熔断器配置
type Config struct {
	ErrorRatio       float64       // 错误率阈值，超过此值触发熔断（如 0.5 = 50%）
	MinRequests      int           // 最小请求数，低于此数不触发熔断
	PassDuration     time.Duration // 半开后等待多久放试探请求（如 30s）
	TripeRequests    int           // 半开状态下允许的试探请求数（如 3）
	SuccessThreshold int           // 半开状态下需要多少成功才算恢复
}

// CircuitBreaker 熔断器
type CircuitBreaker struct {
	config           Config
	state            atomic.Int64 // State
	failedCount      atomic.Int64 // 当前窗口失败次数
	successCount     atomic.Int64 // 当前窗口成功次数
	totalCount       atomic.Int64 // 当前窗口总请求数
	lastFailureTime  atomic.Int64 // 最后一次失败的时间戳 (Unix nano)
	halfOpenSuccess  atomic.Int64 // 半开状态下的连续成功次数
	halfOpenWindow   time.Time    // 半开窗口开启时间
	mu               sync.Mutex   // 状态切换保护锁
}

// 步骤 1：根据当前状态决定是否放行请求
func (cb *CircuitBreaker) AllowRequest() bool {
	currentState := State(cb.state.Load())

	switch currentState {
	case StateClosed:
		// 闭合状态：放行请求，统计结果
		cb.totalCount.Add(1)
		return true

	case StateOpen:
		// 步骤 2：打开状态，检查是否过了恢复等待期
		lastFail := time.UnixNano(cb.lastFailureTime.Load())
		if time.Since(lastFail) > cb.config.PassDuration {
			// 切换到半开
			cb.state.Store(int64(StateHalfOpen))
			cb.halfOpenSuccess.Store(0)
			klog.Warnf("circuit breaker: OPEN → HALF_OPEN, allowing trial requests")
			return true
		}
		// 仍在熔断期，快速拒绝
		return false

	case StateHalfOpen:
		// 步骤 3：半开状态，只允许有限的试探请求
		return cb.totalCount.Load()-cb.halfOpenSuccess.Load() <= int64(cb.config.TripeRequests)

	default:
		return false
	}
}

// 步骤 4：记录成功，半开态下达到阈值则恢复闭合
func (cb *CircuitBreaker) RecordSuccess() {
	currentState := State(cb.state.Load())
	cb.totalCount.Add(1)
	cb.successCount.Add(1)

	if currentState == StateHalfOpen {
		cb.halfOpenSuccess.Add(1)
		// 半开状态下达到成功阈值 → 恢复闭合
		if cb.halfOpenSuccess.Load() >= int64(cb.config.SuccessThreshold) {
			cb.state.Store(int64(StateClosed))
			cb.resetCounters()
			klog.Info("circuit breaker: HALF_OPEN → CLOSED, service recovered")
		}
	}
}

// 步骤 5：记录失败，半开态失败则重开，闭合态超阈值则熔断
func (cb *CircuitBreaker) RecordFailure() {
	currentState := State(cb.state.Load())
	cb.totalCount.Add(1)
	cb.failedCount.Add(1)
	cb.lastFailureTime.Store(time.Now().UnixNano())

	if currentState == StateHalfOpen {
		// 半开状态下失败 → 重新打开
		cb.state.Store(int64(StateOpen))
		klog.Warn("circuit breaker: HALF_OPEN → OPEN, trial failed, re-opening")
		return
	}

	// 闭合状态下检查是否触发熔断
	if cb.shouldTrip() {
		cb.state.Store(int64(StateOpen))
		klog.Warnf("circuit breaker: CLOSED → OPEN, error ratio %.2f exceeded threshold %.2f",
			float64(cb.failedCount.Load())/float64(cb.totalCount.Load()),
			cb.config.ErrorRatio)
	}
}

// shouldTrip 判断是否触发熔断：样本不足 + 错误率超阈值
func (cb *CircuitBreaker) shouldTrip() bool {
	total := cb.totalCount.Load()
	if total < int64(cb.config.MinRequests) {
		return false
	}
	errorRatio := float64(cb.failedCount.Load()) / float64(total)
	return errorRatio >= cb.config.ErrorRatio
}

func (cb *CircuitBreaker) resetCounters() {
	cb.failedCount.Store(0)
	cb.successCount.Store(0)
	cb.totalCount.Store(0)
}

// 步骤 1：检查是否允许请求（熔断打开则快速失败）
func CircuitBreakerMW(cb *CircuitBreaker) endpoint.Middleware {
	return func(next endpoint.Endpoint) endpoint.Endpoint {
		return func(ctx context.Context, req, resp interface{}) (err error) {
			// 步骤 2：熔断检查
			if !cb.AllowRequest() {
				klog.Warn("circuit breaker: OPEN, request rejected (fast-fail)")
				return &CircuitBreakerOpenError{msg: "circuit breaker open"}
			}

			// 步骤 3：执行实际调用
			err = next(ctx, req, resp)

			// 步骤 4：根据结果记录成功或失败
			if err != nil {
				cb.RecordFailure()
			} else {
				cb.RecordSuccess()
			}

			return err
		}
	}
}

// CircuitBreakerOpenError 熔断打开时的错误
type CircuitBreakerOpenError struct{ msg string }
func (e *CircuitBreakerOpenError) Error() string   { return e.msg }
func (e *CircuitBreakerOpenError) Is(target error) bool {
	_, ok := target.(*CircuitBreakerOpenError)
	return ok
}

// 步骤 1：创建熔断器
cb := &CircuitBreaker{
	config: Config{
		ErrorRatio:       0.5,       // 错误率超过 50% 触发
		MinRequests:      20,        // 至少 20 个请求才判定
		PassDuration:     30 * time.Second, // 30s 后进入半开
		TripeRequests:    3,         // 半开允许 3 个试探
		SuccessThreshold: 2,         // 2 个成功即恢复
	},
}

// 步骤 2：客户端注入中间件
client, err := math.NewClient(
	"dqq.math",
	client.WithMiddleware(CircuitBreakerMW(cb)),
	client.WithRPCTimeout(200*time.Millisecond),
)

go get github.com/alibaba/sentinel-golang/pkg/core/config
go get github.com/alibaba/sentinel-golang/pkg adapter

import (
	"github.com/alibaba/sentinel-golang/core/circuitbreaker"
	"github.com/alibaba/sentinel-golang/core/system"
	"github.com/alibaba/sentinel-golang/adapters/kitex"
)

// 步骤 1：初始化 Sentinel，指定熔断策略
cfg := config.NewDefaultConfig()
cfg.Sentinel.CircuitBreakerStrategy = circuitbreaker.StrategyErrorCountRatio // 或 ErrorRatio
config.ApplyConfig(cfg)

// 步骤 2：定义熔断规则
rule := circuitbreaker.Rule{
	Resource:                 "dqq.math/Service.Sub",
	Strategy:                 circuitbreaker.StrategyErrorRatio,
	RequestCount:             20,           // 统计窗口内的请求数
	ErrorRatioThreshold:      0.5,          // 错误率 50%
	MinRequestNumber:         10,           // 最小请求数
	SlowRatioThreshold:       0.0,          // 慢调用比例（可选）
	MaxAllowedSlows:          0,            // 半开期间最大慢调用数
	StatIntervalMs:           10000,        // 统计窗口 10s
	Temperature:              0,            // 温度因子
	RecoveryTimeoutMs:        30000,        // 恢复超时 30s
	VarianceCoefficient:      1.0,          // 方差系数
	MeanSlowRequestDuration:  0,            // 慢调用阈值
}
// 步骤 3：注册规则
circuitbreaker.PutRule(rule)

import "github.com/alibaba/sentinel-golang/adapters/kitex"

// 只需注入 Sentinel 提供的中间件即可，无需自己维护状态机
client, err := math.NewClient(
	"dqq.math",
	client.WithMiddleware(kitex.ClientSentinelMiddleware()),
	client.WithRPCTimeout(200*time.Millisecond),
)

// 步骤 1：定义滑动窗口结构（多个原子计数器桶）
type SlidingWindowCounter struct {
	interval     time.Duration
	buckets      []atomic.Int64
	bucketCount  int
 currentIndex int
}

// 步骤 2：初始化所有桶为 0
func NewSlidingWindowCounter(interval time.Duration, buckets int) *SlidingWindowCounter {
	arr := make([]atomic.Int64, buckets)
	for i := range arr {
		arr[i].Store(0)
	}
	return &SlidingWindowCounter{
		interval:    interval,
		buckets:     arr,
		bucketCount: buckets,
	}
}

// 步骤 3：根据当前时间戳定位桶下标（取模实现环形复用）
func (sw *SlidingWindowCounter) currentBucketIndex() int {
	return int(time.Now().UnixNano() / sw.interval.Nanoseconds()) % sw.bucketCount
}

// 步骤 4：计数。成功 +1，失败 -1（负数表达失败，便于统一统计）
func (sw *SlidingWindowCounter) Increment(isSuccess bool) {
	idx := sw.currentBucketIndex()
	if isSuccess {
		sw.buckets[idx].Add(1)
	} else {
		sw.buckets[idx].Add(-1) // 负数表示失败
	}
}

// 步骤 5：遍历所有桶计算错误率
func (sw *SlidingWindowCounter) ErrorRatio() (float64, int) {
	var total, failed int64
	for _, b := range sw.buckets {
		v := b.Load()
		if v > 0 {
			total += v
		} else {
			failed += -v
			total += v // v 是负数
		}
	}
	if total == 0 {
		return 0, 0
	}
	return float64(failed) / float64(total), int(total)
}

Config{
    // 错误率阈值：建议 0.4 ~ 0.6
    // 太低容易误熔断，太高起不到保护作用
    ErrorRatio: 0.5,

    // 最小请求数：建议 10 ~ 30
    // 避免样本太少时误判
    MinRequests: 20,

    // 恢复等待时间：建议 15s ~ 60s
    // 给下游足够的恢复时间
    PassDuration: 30 * time.Second,

    // 半开试探请求数：建议 2 ~ 5
    // 太少无法判断，太多可能压垮正在恢复的下游
    TripeRequests: 3,

    // 半开成功阈值：建议 50% ~ 80% 的试探请求
    SuccessThreshold: 2,
}

// 步骤 1：熔断打开时不再快速失败，而是走降级逻辑
func CircuitBreakerWithFallback(cb *CircuitBreaker) endpoint.Middleware {
	return func(next endpoint.Endpoint) endpoint.Endpoint {
		return func(ctx context.Context, req, resp interface{}) (err error) {
			if !cb.AllowRequest() {
				// 熔断打开时执行降级逻辑
				return doFallback(req)
			}

			// 步骤 2：正常路径，仍要记录结果
			err = next(ctx, req, resp)
			if err != nil {
				cb.RecordFailure()
			} else {
				cb.RecordSuccess()
			}
			return err
		}
	}
}

// 步骤 3：降级返回缓存数据、默认值或友好提示
func doFallback(req interface{}) error {
	// 返回缓存数据、默认值、或友好提示
	return &FallbackError{msg: "服务暂时不可用，请稍后重试"}
}

// 数学计算服务：Sub 可以熔断，Health 不应该熔断
cbMap := map[string]*CircuitBreaker{
	"Sub":       {config: Config{ErrorRatio: 0.5, MinRequests: 20}},
	"ComplexCalc": {config: Config{ErrorRatio: 0.3, MinRequests: 10}},
}

func PerMethodCBMW(cbMap map[string]*CircuitBreaker) endpoint.Middleware {
	return func(next endpoint.Endpoint) endpoint.Endpoint {
		return func(ctx context.Context, req, resp interface{}) (err error) {
			method := getMethodName(ctx) // 从 ctx 提取方法名
			cb, ok := cbMap[method]
			if !ok {
				return next(ctx, req, resp)
			}
			// ... 同上的熔断逻辑
		}
	}
}

// 在状态切换时上报指标
func (cb *CircuitBreaker) transition(from, to State) {
	klog.Warnf("[CIRCUIT_BREAKER] service=%s state=%s→%s",
		serviceName, from, to)
	// 上报到 Prometheus / StatsD
	metrics.Counter("circuit_breaker_state_change",
		prometheus.Labels{"service": serviceName, "from": from.String(), "to": to.String()},
		1)
}