Browse Source

feat: change requestTimeout proto define sequence for old compatible and optimize grpc interceptor set request timeout

pull/228/head
liulei 4 years ago
parent
commit
c507acee37
  1. 42
      dtmgrpc/dtmgpb/dtmgimp.pb.go
  2. 6
      dtmgrpc/dtmgpb/dtmgimp.proto
  3. 12
      dtmsvr/svr.go
  4. 17
      dtmsvr/trans_status.go

42
dtmgrpc/dtmgpb/dtmgimp.pb.go

@ -29,9 +29,9 @@ type DtmTransOptions struct {
WaitResult bool `protobuf:"varint,1,opt,name=WaitResult,proto3" json:"WaitResult,omitempty"`
TimeoutToFail int64 `protobuf:"varint,2,opt,name=TimeoutToFail,proto3" json:"TimeoutToFail,omitempty"`
RetryInterval int64 `protobuf:"varint,3,opt,name=RetryInterval,proto3" json:"RetryInterval,omitempty"`
RequestTimeout int64 `protobuf:"varint,4,opt,name=RequestTimeout,proto3" json:"RequestTimeout,omitempty"`
PassthroughHeaders []string `protobuf:"bytes,5,rep,name=PassthroughHeaders,proto3" json:"PassthroughHeaders,omitempty"`
BranchHeaders map[string]string `protobuf:"bytes,6,rep,name=BranchHeaders,proto3" json:"BranchHeaders,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
PassthroughHeaders []string `protobuf:"bytes,4,rep,name=PassthroughHeaders,proto3" json:"PassthroughHeaders,omitempty"`
BranchHeaders map[string]string `protobuf:"bytes,5,rep,name=BranchHeaders,proto3" json:"BranchHeaders,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
RequestTimeout int64 `protobuf:"varint,6,opt,name=RequestTimeout,proto3" json:"RequestTimeout,omitempty"`
}
func (x *DtmTransOptions) Reset() {
@ -87,13 +87,6 @@ func (x *DtmTransOptions) GetRetryInterval() int64 {
return 0
}
func (x *DtmTransOptions) GetRequestTimeout() int64 {
if x != nil {
return x.RequestTimeout
}
return 0
}
func (x *DtmTransOptions) GetPassthroughHeaders() []string {
if x != nil {
return x.PassthroughHeaders
@ -108,6 +101,13 @@ func (x *DtmTransOptions) GetBranchHeaders() map[string]string {
return nil
}
func (x *DtmTransOptions) GetRequestTimeout() int64 {
if x != nil {
return x.RequestTimeout
}
return 0
}
// DtmRequest request sent to dtm server
type DtmRequest struct {
state protoimpl.MessageState
@ -353,17 +353,17 @@ var file_dtmgrpc_dtmgpb_dtmgimp_proto_rawDesc = []byte{
0x0d, 0x54, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x54, 0x6f, 0x46, 0x61, 0x69, 0x6c, 0x12, 0x24,
0x0a, 0x0d, 0x52, 0x65, 0x74, 0x72, 0x79, 0x49, 0x6e, 0x74, 0x65, 0x72, 0x76, 0x61, 0x6c, 0x18,
0x03, 0x20, 0x01, 0x28, 0x03, 0x52, 0x0d, 0x52, 0x65, 0x74, 0x72, 0x79, 0x49, 0x6e, 0x74, 0x65,
0x72, 0x76, 0x61, 0x6c, 0x12, 0x26, 0x0a, 0x0e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54,
0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x18, 0x04, 0x20, 0x01, 0x28, 0x03, 0x52, 0x0e, 0x52, 0x65,
0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x12, 0x2e, 0x0a, 0x12,
0x50, 0x61, 0x73, 0x73, 0x74, 0x68, 0x72, 0x6f, 0x75, 0x67, 0x68, 0x48, 0x65, 0x61, 0x64, 0x65,
0x72, 0x73, 0x18, 0x05, 0x20, 0x03, 0x28, 0x09, 0x52, 0x12, 0x50, 0x61, 0x73, 0x73, 0x74, 0x68,
0x72, 0x6f, 0x75, 0x67, 0x68, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x73, 0x12, 0x51, 0x0a, 0x0d,
0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x73, 0x18, 0x06, 0x20,
0x03, 0x28, 0x0b, 0x32, 0x2b, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x69, 0x6d, 0x70, 0x2e, 0x44, 0x74,
0x6d, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x2e, 0x42, 0x72,
0x61, 0x6e, 0x63, 0x68, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79,
0x52, 0x0d, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x73, 0x1a,
0x72, 0x76, 0x61, 0x6c, 0x12, 0x2e, 0x0a, 0x12, 0x50, 0x61, 0x73, 0x73, 0x74, 0x68, 0x72, 0x6f,
0x75, 0x67, 0x68, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x73, 0x18, 0x04, 0x20, 0x03, 0x28, 0x09,
0x52, 0x12, 0x50, 0x61, 0x73, 0x73, 0x74, 0x68, 0x72, 0x6f, 0x75, 0x67, 0x68, 0x48, 0x65, 0x61,
0x64, 0x65, 0x72, 0x73, 0x12, 0x51, 0x0a, 0x0d, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x48, 0x65,
0x61, 0x64, 0x65, 0x72, 0x73, 0x18, 0x05, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x2b, 0x2e, 0x64, 0x74,
0x6d, 0x67, 0x69, 0x6d, 0x70, 0x2e, 0x44, 0x74, 0x6d, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x4f, 0x70,
0x74, 0x69, 0x6f, 0x6e, 0x73, 0x2e, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x48, 0x65, 0x61, 0x64,
0x65, 0x72, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x0d, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68,
0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x73, 0x12, 0x26, 0x0a, 0x0e, 0x52, 0x65, 0x71, 0x75, 0x65,
0x73, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x18, 0x06, 0x20, 0x01, 0x28, 0x03, 0x52,
0x0e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x1a,
0x40, 0x0a, 0x12, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x73,
0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01,
0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65,

6
dtmgrpc/dtmgpb/dtmgimp.proto

@ -18,9 +18,9 @@ message DtmTransOptions {
bool WaitResult = 1;
int64 TimeoutToFail = 2;
int64 RetryInterval = 3;
int64 RequestTimeout = 4;
repeated string PassthroughHeaders = 5;
map<string, string> BranchHeaders = 6;
repeated string PassthroughHeaders = 4;
map<string, string> BranchHeaders = 5;
int64 RequestTimeout = 6;
}
// DtmRequest request sent to dtm server

12
dtmsvr/svr.go

@ -7,14 +7,12 @@
package dtmsvr
import (
"context"
"fmt"
"net"
"time"
"github.com/dtm-labs/dtm/dtmcli"
"github.com/dtm-labs/dtm/dtmcli/logger"
"github.com/dtm-labs/dtm/dtmgrpc"
"github.com/dtm-labs/dtm/dtmgrpc/dtmgimp"
"github.com/dtm-labs/dtm/dtmgrpc/dtmgpb"
"github.com/dtm-labs/dtm/dtmutil"
@ -28,11 +26,11 @@ func StartSvr() {
setServerInfoMetrics()
dtmcli.GetRestyClient().SetTimeout(time.Duration(conf.RequestTimeout) * time.Second)
dtmgrpc.AddUnaryInterceptor(func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
ctx2, cancel := context.WithTimeout(ctx, time.Duration(conf.RequestTimeout)*time.Second)
defer cancel()
return invoker(ctx2, method, req, reply, cc, opts...)
})
//dtmgrpc.AddUnaryInterceptor(func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
// ctx2, cancel := context.WithTimeout(ctx, time.Duration(conf.RequestTimeout)*time.Second)
// defer cancel()
// return invoker(ctx2, method, req, reply, cc, opts...)
//})
// start gin server
app := dtmutil.GetGinApp()

17
dtmsvr/trans_status.go

@ -122,14 +122,15 @@ func (t *TransGlobal) getURLResult(url string, branchID, op string, branchPayloa
if err != nil {
return err
}
if t.RequestTimeout != 0 {
// use ClientInterceptors[1:] for remove default request setting
dtmgimp.ClientInterceptors = append(dtmgimp.ClientInterceptors[1:], func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
ctx2, cancel := context.WithTimeout(ctx, time.Duration(t.RequestTimeout)*time.Second)
defer cancel()
return invoker(ctx2, method, req, reply, cc, opts...)
})
}
dtmgimp.ClientInterceptors = append(dtmgimp.ClientInterceptors, func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
timeout := conf.RequestTimeout
if t.RequestTimeout != 0 {
timeout = conf.RequestTimeout
}
ctx2, cancel := context.WithTimeout(ctx, time.Duration(timeout)*time.Second)
defer cancel()
return invoker(ctx2, method, req, reply, cc, opts...)
})
conn := dtmgimp.MustGetGrpcConn(server, true)
ctx := dtmgimp.TransInfo2Ctx(t.Gid, t.TransType, branchID, op, "")
kvs := dtmgimp.Map2Kvs(t.Ext.Headers)

Loading…
Cancel
Save