sentinel go

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

func MakeGrpcServerInterceptor() func(context.Context, interface{}, *grpc.UnaryServerInfo, grpc.UnaryHandler) (interface{}, error) {
	sentinel.Logger.Info("sentinel breaker server is open")
	ruleLoadMap := sync.Map{}
	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {

		_scope := "server.grpc." + info.FullMethod
		_, exists := ruleLoadMap.Load(_scope)
		//如果该规则不存在，则说明是第一次请求，则需要加载规则,加载规则在并发时可以多次，反正是覆盖，不影响。
		if !exists {
			rules := getRuleByScope(strings.Split(_scope, "."))
			circuitbreaker.LoadRulesOfResource(_scope, rules)
			ruleLoadMap.Store(_scope, "1")
		}

		token, e := api.Entry(_scope, api.WithTrafficType(base.Inbound))
		if e != nil {
			sentinel.Logger.ContextErrorWithFields(ctx, logrus.Fields{
				"err":    "grpc接口熔断",
				"rule":   e.TriggeredRule(),
				"_scope": _scope,
			})
			return nil, error2.SysErrSentinelBreaker
		}
		defer token.Exit()
		res, err := handler(ctx, req)
		if err != nil {
			e := error2.ParseErrToNeoErr(err)
			//只有系统错误才算 而且如果是外部资源熔断或者上游熔断返回的错误，是不触发下游节点的熔断的
			if e.Code < error2.MaxSystemErr && e.Code != error2.ResourceErrSentinelBreaker.Code && e.Code != error2.SysErrSentinelBreaker.Code {
				api.TraceError(token, e)
			}
		}
		return res, err
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

// scope 作用域，指定熔断器的基础设置，
// scope采用三层，即 type.name.method
// type代表大的种类，比如 server/mongo/mysql/redis/grpc/http
// name代表具体的实例 比如当type=mongo时，name代表collection 如user post等
// method代表具体的方法或表，即熔断的最小单位， 如果是http，则method可以是具体的方法， 如果是mongo，则method是具体的语句 如：find findOne
func Middleware(endpoint endpoint.Endpoint) endpoint.Endpoint {
	//scopes := strings.Split(scope, ".")
	//if len(scopes) != 3 {
	//	panic("熔断组件层级必须是3")
	//}
	//rules := getRuleByScope(scopes)
	//circuitbreaker.LoadRulesOfResource(scope, rules)

	ruleLoadMap := sync.Map{}
	return func(ctx context.Context, request interface{}) (response interface{}, err error) {
		_scope := getScopeByCtx(ctx)
		_, exists := ruleLoadMap.Load(_scope)
		//如果该规则不存在，则说明是第一次请求，则需要加载规则,加载规则在并发时可以多次，反正是覆盖，不影响。
		if !exists {
			rules := getRuleByScope(strings.Split(_scope, "."))
			circuitbreaker.LoadRulesOfResource(_scope, rules)
			ruleLoadMap.Store(_scope, "1")
		}
		token, e := api.Entry(_scope, api.WithTrafficType(base.Outbound))
		if e != nil {
			sentinel.Logger.ContextErrorWithFields(ctx, logrus.Fields{
				"err":    "外部资源熔断",
				"rule":   e.TriggeredRule(),
				"_scope": _scope,
			})
			return nil, error2.ResourceErrSentinelBreaker
		}
		defer token.Exit()
		res, err := endpoint(ctx, request)
		if err != nil {
			scopes := strings.Split(_scope, ".")
			//如果是mongo，则对执行耗时超过最大限制的请求标记为错误
			if scopes[0] == Type_Mongo && err.Error() == _mongo_max_time_limit_err {
				api.TraceError(token, e)
			}
			//TODO: 陆续还要支持其它外部资源的熔断
		}
		return res, err
	}
}

1
2

// 定义资源名称
resName := "test-resource"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// 定义流控规则
flowRule := &flow.Rule{
    // 熔断的异常比例
    Count: 10,
    // 统计时间窗口，单位为秒
    StatIntervalInMs: 5000,
}
// 定义熔断规则
circuitBreakerRule := &circuitbreaker.Rule{
    // 熔断器的最小请求数量
    MinRequestAmount: 10,
    // 熔断器的统计时间窗口，单位为秒
    StatIntervalMs: 5000,
    // 熔断器的失败比例阈值
    ErrorThresholdPercentage: 50,
}

1
2
3
4
5
6
7
8

// 获取资源对象
res := flow.GetResource(resName)
// 注册流控规则
if res != nil {
    flow.LoadRules([]*flow.Rule{flowRule})
}
// 注册熔断规则
circuitbreaker.LoadRules([]*circuitbreaker.Rule{circuitBreakerRule})

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

// 获取资源统计信息
res := flow.GetResource(resName)
// 统计流量
if res != nil {
    // 记录QPS
    flow.IncreaseQPS(resName, 1)
    // 监控流量
    if flow.IsAllowed(resName, 1) {
        // 处理请求
        handleRequest()
    } else {
        // 触发熔断保护
        handleCircuitBreaker()
    }
}