timeout 无法触发breaker

[LeeDF]

模拟timeout场景, 无法触发breaker

fooService.proto

syntax = "proto3";


package fooService;


message sleepRequest {
  int64 sleep = 1;
}

message sleepResponse {
  string msg = 2;
}



service fooService {

  rpc sleep(sleepRequest) returns (sleepResponse);
}

sleeplogic.go

package logic

import (
	"context"
	"time"

	"github.com/tal-tech/go-zero/core/logx"
	"gitlab.excelliance.cn/zm2c-ourplay/apiservice/rpc/fooService/fooService"
	"gitlab.excelliance.cn/zm2c-ourplay/apiservice/rpc/fooService/internal/svc"
)

type SleepLogic struct {
	ctx    context.Context
	svcCtx *svc.ServiceContext
	logx.Logger
}

func NewSleepLogic(ctx context.Context, svcCtx *svc.ServiceContext) *SleepLogic {
	return &SleepLogic{
		ctx:    ctx,
		svcCtx: svcCtx,
		Logger: logx.WithContext(ctx),
	}
}

func (l *SleepLogic) Sleep(in *fooService.SleepRequest) (*fooService.SleepResponse, error) {
	if in.Sleep > 0 {
		time.Sleep(time.Second * time.Duration(in.Sleep))
	}

	return &fooService.SleepResponse{Msg: "sleep"}, nil
}

fooservice.yaml

 Name: fooservice
ListenOn: 127.0.0.1:12345
Timeout: 3000
Etcd:
  Hosts:
    - 127.0.0.1:2379
  Key: fooservice.op
Rpc:
  Timeout: 10000

模拟调用

l.svcCtx.FooServiceRpc.Sleep(l.ctx, &fooService.SleepRequest{CommonIn: cin, Sleep: int64(req.SleepTime)})

发现大量调用下不会出发breaker, debug发现 err 返回的是rpc error: code = Unknown desc = context deadline exceeded
unknown code会 通过breaker标记成功

func (b *googleBreaker) doReq(req func() error, fallback func(err error) error, acceptable Acceptable) error {
	if err := b.accept(); err != nil {
		if fallback != nil {
			return fallback(err)
		}

		return err
	}

	defer func() {
		if e := recover(); e != nil {
			b.markFailure()
			panic(e)
		}
	}()

	err := req()
//rpc error: code = Unknown desc = context deadline exceeded
	if acceptable(err) {
//success
		b.markSuccess()
	} else {
		b.markFailure()
	}

	return err
}

[LeeDF]

修改timeoutinterceptor.go可修复此问题
// UnaryTimeoutInterceptor returns a func that sets timeout to incoming unary requests.
func UnaryTimeoutInterceptor(timeout time.Duration) grpc.UnaryServerInterceptor {
	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo,
		handler grpc.UnaryHandler) (interface{}, error) {
		ctx, cancel := context.WithTimeout(ctx, timeout)
		defer cancel()

		var resp interface{}
		var err error
		var lock sync.Mutex
		done := make(chan struct{})
		// create channel with buffer size 1 to avoid goroutine leak
		panicChan := make(chan interface{}, 1)
		go func() {
			defer func() {
				if p := recover(); p != nil {
					panicChan <- p
				}
			}()

			lock.Lock()
			defer lock.Unlock()
			resp, err = handler(ctx, req)
			close(done)
		}()

		select {
		case p := <-panicChan:
			panic(p)
		case <-done:
			lock.Lock()
			defer lock.Unlock()
			return resp, err
		case <-ctx.Done():
                        //修改此处
			return nil, status.FromContextError(ctx.Err()).Err()
			//return nil, ctx.Err()
		}
	}
}