Optimize timeout statusCode (#521)

* optimize timeout statusCode

* remove chinese

* feat : remove dubbo-go-pixiu-samples

Co-authored-by: mark4z <mark4z@apache.org>

pixiu/pkg/client/dubbo/dubbo.go

@@ -210,6 +210,7 @@ func (dc *Client) Call(req *client.Request) (res interface{}, err error) {
 	ctx := context.WithValue(req.Context, constant.TracingRemoteSpanCtx, trace.SpanFromContext(req.Context).SpanContext())
 	rst, err := gs.Invoke(ctx, method, types, vals)
 	if err != nil {
+		// TODO statusCode I don’t know what dubbo will return when it times out, so I will return it directly. I will judge it when I call it.
 		return nil, err
 	}
 

pixiu/pkg/client/http/http.go

@@ -122,6 +122,10 @@ func (dc *Client) Call(req *client.Request) (resp interface{}, err error) {
 	}
 	if err != nil {
 		span.AddEvent(semconv.ExceptionEventName, trace.WithAttributes(semconv.ExceptionMessageKey.String(err.Error())))
+		urlErr, ok := err.(*url.Error)
+		if ok && urlErr.Timeout() {
+			err = errors.Errorf("http req call timeout err: %s", err.Error())
+		}
 	}
 
 	return tmpRet, err

pixiu/pkg/client/triple/triple.go

@@ -26,9 +26,11 @@ import (
 )
 
 import (
+	gerrors "github.com/mercari/grpc-http-proxy/errors"
 	proxymeta "github.com/mercari/grpc-http-proxy/metadata"
 	"github.com/mercari/grpc-http-proxy/proxy"
 	"github.com/pkg/errors"
+	"google.golang.org/grpc/codes"
 )
 
 import (
@@ -92,6 +94,13 @@ func (dc *Client) Call(req *client.Request) (res interface{}, err error) {
 	defer cancel()
 	call, err := p.Call(ctx, req.API.Method.IntegrationRequest.Interface, req.API.Method.IntegrationRequest.Method, reqData, (*proxymeta.Metadata)(&meta))
 	if err != nil {
+		gerr, ok := err.(*gerrors.GRPCError)
+		if ok {
+			statusCode := codes.Code(gerr.StatusCode)
+			if statusCode == codes.Canceled || statusCode == codes.DeadlineExceeded {
+				return "", errors.Errorf("call triple server timeout error = %s", err)
+			}
+		}
 		return "", errors.Errorf("call triple server error = %s", err)
 	}
 	return call, nil

pixiu/pkg/common/grpc/manager.go

@@ -90,6 +90,10 @@ func (gcm *GrpcConnectionManager) ServeHTTP(w stdHttp.ResponseWriter, r *stdHttp
 
 	if err != nil {
 		logger.Infof("GrpcConnectionManager forward request error %v", err)
+		if err == context.DeadlineExceeded {
+			gcm.writeStatus(w, status.New(codes.DeadlineExceeded, fmt.Sprintf("forward timeout error = %v", err)))
+			return
+		}
 		gcm.writeStatus(w, status.New(codes.Unknown, fmt.Sprintf("forward error not = %v", err)))
 		return
 	}

pixiu/pkg/filter/http/dubboproxy/dubbo.go

@@ -217,13 +217,18 @@ func (f *Filter) Decode(hc *pixiuHttp.HttpContext) filter.FilterStatus {
 	var resp interface{}
 	invoc.SetReply(&resp)
 
-	invCtx := context.Background()
+	invCtx, cancel := context.WithTimeout(context.Background(), hc.Timeout)
+	defer cancel()
 	result := invoker.Invoke(invCtx, invoc)
 	result.SetAttachments(invoc.Attachments())
 
 	if result.Error() != nil {
 		logger.Debugf("[dubbo-go-pixiu] invoke result error %v", result.Error())
 		bt, _ := json.Marshal(pixiuHttp.ErrResponse{Message: fmt.Sprintf("invoke result error %v", result.Error())})
+		// TODO statusCode I don't know what dubbo returns when it times out, first use the string to judge
+		if strings.Contains(result.Error().Error(), "timeout") {
+			hc.SendLocalReply(http.StatusGatewayTimeout, bt)
+		}
 		hc.SendLocalReply(http.StatusServiceUnavailable, bt)
 		return filter.Stop
 	}

pixiu/pkg/filter/http/grpcproxy/grpc.go

@@ -265,6 +265,11 @@ func (f *Filter) Decode(c *http.HttpContext) filter.FilterStatus {
 	resp, err := Invoke(ctx, stub, mthDesc, grpcReq, grpc.Header(&md), grpc.Trailer(&t))
 	// judge err is server side error or not
 	if st, ok := status.FromError(err); !ok || isServerError(st) {
+		if isServerTimeout(st) {
+			logger.Errorf("%s err {failed to invoke grpc service provider because timeout, err:%s}", loggerHeader, err.Error())
+			c.SendLocalReply(stdHttp.StatusGatewayTimeout, []byte(fmt.Sprintf("%s", err)))
+			return filter.Stop
+		}
 		logger.Errorf("%s err {failed to invoke grpc service provider, %s}", loggerHeader, err.Error())
 		c.SendLocalReply(stdHttp.StatusServiceUnavailable, []byte(fmt.Sprintf("%s", err)))
 		return filter.Stop
@@ -371,6 +376,10 @@ func isServerError(st *status.Status) bool {
 		st.Code() == codes.Unavailable
 }
 
+func isServerTimeout(st *status.Status) bool {
+	return st.Code() == codes.DeadlineExceeded || st.Code() == codes.Canceled
+}
+
 func (factory *FilterFactory) Config() interface{} {
 	return factory.cfg
 }

pixiu/pkg/filter/http/httpproxy/routerfilter.go

@@ -126,6 +126,11 @@ func (f *Filter) Decode(hc *http.HttpContext) filter.FilterStatus {
 
 	resp, err := cli.Do(req)
 	if err != nil {
+		urlErr, ok := err.(*url.Error)
+		if ok && urlErr.Timeout() {
+			hc.SendLocalReply(http3.StatusGatewayTimeout, []byte(err.Error()))
+			return filter.Stop
+		}
 		hc.SendLocalReply(http3.StatusServiceUnavailable, []byte(err.Error()))
 		return filter.Stop
 	}

pixiu/pkg/filter/http/remote/call.go

@@ -143,6 +143,10 @@ func (f *Filter) Decode(c *contexthttp.HttpContext) filter.FilterStatus {
 	resp, err := cli.Call(req)
 	if err != nil {
 		logger.Errorf("[dubbo-go-pixiu] client call err:%v!", err)
+		if strings.Contains(strings.ToLower(err.Error()), "timeout") {
+			c.SendLocalReply(http.StatusGatewayTimeout, []byte(fmt.Sprintf("client call timeout err: %s", err)))
+			return filter.Stop
+		}
 		c.SendLocalReply(http.StatusInternalServerError, []byte(fmt.Sprintf("client call err: %s", err)))
 		return filter.Stop
 	}