github/go-gin-api
4
0
Fork 0
mirror of https://github.com/xinliangnote/go-gin-api.git synced 2024-04-21 12:31:46 +00:00
Code Issues Releases Wiki Activity

upgrade

Browse Source
This commit is contained in:
新亮
2021-02-09 09:52:50 +08:00
parent a4701309cb
commit 334bfbb4aa
12 changed files with 263 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/controller/demo/demo.go
+7 -24
View File
@@ -10,6 +10,7 @@ import (
"github.com/xinliangnote/go-gin-api/internal/pkg/cache"
"github.com/xinliangnote/go-gin-api/internal/pkg/core"
"github.com/xinliangnote/go-gin-api/internal/pkg/db"
"github.com/xinliangnote/go-gin-api/internal/pkg/grpc"
"github.com/xinliangnote/go-gin-api/pkg/httpclient"
"github.com/xinliangnote/go-gin-api/pkg/p"
"github.com/xinliangnote/go-gin-api/pkg/token"
@@ -21,13 +22,15 @@ import (
type Demo struct {
logger *zap.Logger
cache cache.Repo
grpconn grpc.ClientConn
userService user_service.UserService
}
func NewDemo(logger *zap.Logger, db db.Repo, cache cache.Repo) *Demo {
func NewDemo(logger *zap.Logger, db db.Repo, cache cache.Repo, grpconn grpc.ClientConn) *Demo {
return &Demo{
logger: logger,
cache: cache,
grpconn: grpconn,
userService: user_service.NewUserService(db, cache),
}
}
@@ -184,29 +187,9 @@ func (d *Demo) Trace() core.HandlerFunc {
val, _ := d.cache.Get("name", cache.WithTrace(c.Trace()))
p.Println("redis-name", val, p.WithTrace(c.Trace()))
//// 执行 gRPC 信息
//conn, err := grpclient.New("127.0.0.1:9001",
// grpclient.WithDialTimeout(time.Second*5),
// grpclient.WithTrace(c.Trace()),
// grpclient.WithSign(func(message []byte) (authorization string, err error) {
// return grpclient.GenerateSign("abcdef", message)
// }),
//)
//if err != nil {
// d.logger.Error("grpc conn err", zap.Error(err))
//}
//defer conn.Close()
//
//// 初始化客户端
//client := hello.NewHelloClient(conn)
//
//// 调用 SayHello 方法
//ctx := context.Background()
//// 设置 SayHello 超时间
//ctx, cancel := context.WithTimeout(ctx, time.Second*3)
//defer cancel()
//
//client.SayHello(ctx, &hello.HelloRequest{Name: "Hello World"})
// 初始化客户端
// client := hello.NewHelloClient(d.grpconn.Conn())
// client.SayHello(grpc.ContextWithValueAndTimeout(c, time.Second*3), &hello.HelloRequest{Name: "Hello World"})
data := &traceResponse{
{
internal/api/router/router.go
+3 -2
View File
@@ -8,6 +8,7 @@ import (
"github.com/xinliangnote/go-gin-api/internal/pkg/cache"
"github.com/xinliangnote/go-gin-api/internal/pkg/core"
"github.com/xinliangnote/go-gin-api/internal/pkg/db"
"github.com/xinliangnote/go-gin-api/internal/pkg/grpc"
"github.com/xinliangnote/go-gin-api/internal/pkg/metrics"
"github.com/xinliangnote/go-gin-api/internal/pkg/notify"
@@ -15,7 +16,7 @@ import (
"go.uber.org/zap"
)
func NewHTTPMux(logger *zap.Logger, db db.Repo, cache cache.Repo) (core.Mux, error) {
func NewHTTPMux(logger *zap.Logger, db db.Repo, cache cache.Repo, grpconn grpc.ClientConn) (core.Mux, error) {
if logger == nil {
return nil, errors.New("logger required")
@@ -32,7 +33,7 @@ func NewHTTPMux(logger *zap.Logger, db db.Repo, cache cache.Repo) (core.Mux, err
panic(err)
}
demoHandler := demo.NewDemo(logger, db, cache)
demoHandler := demo.NewDemo(logger, db, cache, grpconn)
userHandler := user_handler.NewUserDemo(logger, db, cache)
gqlHandler := handler.New(logger, db, cache)
internal/pkg/cache/redis.go
Vendored
+3 -3
View File
@@ -35,7 +35,7 @@ type Repo interface {
ExpireAt(key string, ttl time.Time) bool
Del(keys ...string) bool
Incr(key string, options ...Option) int64
Close()
Close() error
}
type cacheRepo struct {
@@ -179,8 +179,8 @@ func (c *cacheRepo) Incr(key string, options ...Option) int64 {
}
// Close close redis client
func (c *cacheRepo) Close() {
c.client.Close()
func (c *cacheRepo) Close() error {
return c.client.Close()
}
// WithTrace 设置trace信息
internal/pkg/grpc/grpc.go
+53
View File
@@ -0,0 +1,53 @@
package grpc
import (
"context"
"time"
"google.golang.org/grpc"
)
var _ ClientConn = (*clientConn)(nil)
type ClientConn interface {
i()
Conn() *grpc.ClientConn
}
type clientConn struct {
conn *grpc.ClientConn
}
func New() (ClientConn, error) {
// TODO 需从配置文件中获取
//target := "127.0.0.1:9988"
//secret := "abcdef"
//
//clientInterceptor := NewClientInterceptor(func(message []byte) (authorization string, err error) {
// return GenerateSign(secret, message)
//})
//
//conn, err := grpclient.New(target,
// grpclient.WithKeepAlive(keepAlive),
// grpclient.WithDialTimeout(time.Second*5),
// grpclient.WithUnaryInterceptor(clientInterceptor.UnaryInterceptor),
//)
//
//return &clientConn{
// conn: conn,
//}, err
return nil, nil
}
func (c *clientConn) i() {}
func (c *clientConn) Conn() *grpc.ClientConn {
return c.conn
}
func ContextWithValueAndTimeout(value interface{}, duration time.Duration) context.Context {
ctx, _ := context.WithTimeout(context.Background(), duration)
return context.WithValue(ctx, ClientWithContextKey, value)
}
internal/pkg/grpc/interceptor.go
+142
View File
@@ -0,0 +1,142 @@
package grpc
import (
"context"
"fmt"
"runtime/debug"
"time"
"github.com/xinliangnote/go-gin-api/internal/api/code"
"github.com/xinliangnote/go-gin-api/internal/pkg/core"
"github.com/xinliangnote/go-gin-api/internal/pkg/notify"
"github.com/xinliangnote/go-gin-api/pkg/p"
"github.com/xinliangnote/go-gin-api/pkg/time_parse"
"github.com/xinliangnote/go-gin-api/pkg/trace"
"github.com/golang/protobuf/proto"
"go.uber.org/zap"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
)
const (
// Trace ID
TraceID = "trace-id"
)
type clientWithContextKeyType struct{ name string }
var ClientWithContextKey = clientWithContextKeyType{"_client_with_context"}
// ClientInterceptor the client's interceptor
type ClientInterceptor struct {
sign Sign
}
// NewClientInterceptor create a client interceptor
func NewClientInterceptor(sign Sign) *ClientInterceptor {
return &ClientInterceptor{
sign: sign,
}
}
// UnaryInterceptor a interceptor for client unary operations
func (c *ClientInterceptor) UnaryInterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
var (
invokerErr error
ts = time.Now()
coreContext = ctx.Value(ClientWithContextKey).(core.Context)
)
defer func() { // double recover for safety
if err := recover(); err != nil {
stackInfo := string(debug.Stack())
coreContext.Logger().Error("UnaryInterceptor got double panic", zap.String("panic", fmt.Sprintf("%+v", err)), zap.String("stack", stackInfo))
coreContext.AbortWithError(code.ErrServer)
notify.OnPanicNotify(coreContext, err, stackInfo)
}
}()
defer func() {
if err := recover(); err != nil {
stackInfo := string(debug.Stack())
coreContext.Logger().Error("UnaryInterceptor got panic", zap.String("panic", fmt.Sprintf("%+v", err)), zap.String("stack", stackInfo))
coreContext.AbortWithError(code.ErrServer)
notify.OnPanicNotify(coreContext, err, stackInfo)
}
if coreContext.Trace() != nil {
var mapReq, mapReply map[string]interface{}
mapReq, err := ProtoMessage2Map(req.(proto.Message))
if err != nil {
p.Println("req ProtoMessage2Map err", err, p.WithTrace(coreContext.Trace()))
}
mapReply, err = ProtoMessage2Map(reply.(proto.Message))
if err != nil {
p.Println("reply ProtoMessage2Map err", err, p.WithTrace(coreContext.Trace()))
}
meta, _ := metadata.FromOutgoingContext(ctx)
gRPCTrace := new(trace.Grpc)
gRPCTrace.Timestamp = time_parse.CSTLayoutString()
gRPCTrace.Addr = cc.Target()
gRPCTrace.Method = method
gRPCTrace.Meta = meta
gRPCTrace.Request = mapReq
gRPCTrace.Response = mapReply
gRPCTrace.CostSeconds = time.Since(ts).Seconds()
if invokerErr != nil {
statusErr, ok := status.FromError(invokerErr)
if ok {
gRPCTrace.Code = statusErr.Code().String()
gRPCTrace.Message = statusErr.Message()
}
}
coreContext.Trace().AppendGRPC(gRPCTrace)
}
}()
if c.sign != nil {
var (
raw string
signature string
err error
)
if req != nil {
if raw, err = ProtoMessage2JSON(req.(proto.Message)); err != nil {
return err
}
}
if signature, err = c.sign([]byte(raw)); err != nil {
return err
}
meta, _ := metadata.FromOutgoingContext(ctx)
if meta == nil {
meta = make(metadata.MD)
}
meta.Set(ProxyAuthorization, signature)
ctx = metadata.NewOutgoingContext(ctx, meta)
}
if coreContext.Trace() != nil {
meta, _ := metadata.FromOutgoingContext(ctx)
if meta == nil {
meta = make(metadata.MD)
}
meta.Set(TraceID, coreContext.Trace().ID())
ctx = metadata.NewOutgoingContext(ctx, meta)
}
invokerErr = invoker(ctx, method, req, reply, cc, opts...)
return invokerErr
}
internal/pkg/grpc/keepalive.go
+15
View File
@@ -0,0 +1,15 @@
package grpc
import (
"time"
"google.golang.org/grpc/keepalive"
)
var (
keepAlive = &keepalive.ClientParameters{
Time: 10 * time.Second,
Timeout: time.Second,
PermitWithoutStream: true,
}
)
pkg/grpclient/sign.go → internal/pkg/grpc/sign.go
+1 -1
View File
@@ -1,4 +1,4 @@
package grpclient
package grpc
import (
"bytes"
pkg/grpclient/utils.go → internal/pkg/grpc/utils.go
+1 -1
View File
@@ -1,4 +1,4 @@
package grpclient
package grpc
import (
"bytes"
main.go
+26 -2
View File
@@ -10,6 +10,7 @@ import (
"github.com/xinliangnote/go-gin-api/internal/api/router"
"github.com/xinliangnote/go-gin-api/internal/pkg/cache"
"github.com/xinliangnote/go-gin-api/internal/pkg/db"
"github.com/xinliangnote/go-gin-api/internal/pkg/grpc"
"github.com/xinliangnote/go-gin-api/pkg/env"
"github.com/xinliangnote/go-gin-api/pkg/logger"
"github.com/xinliangnote/go-gin-api/pkg/shutdown"
@@ -54,8 +55,14 @@ func main() {
loggers.Fatal("new cache err", zap.Error(err))
}
// 初始化 gRPC client
gRPCRepo, err := grpc.New()
if err != nil {
loggers.Fatal("new grpc err", zap.Error(err))
}
// 初始化 HTTP 服务
mux, err := router.NewHTTPMux(loggers, dbRepo, cacheRepo)
mux, err := router.NewHTTPMux(loggers, dbRepo, cacheRepo, gRPCRepo)
if err != nil {
panic(err)
}
@@ -89,16 +96,33 @@ func main() {
func() {
if err := dbRepo.DbWClose(); err != nil {
loggers.Error("dbw close err", zap.Error(err))
} else {
loggers.Info("dbw close success")
}
if err := dbRepo.DbRClose(); err != nil {
loggers.Error("dbr close err", zap.Error(err))
} else {
loggers.Info("dbr close success")
}
},
// 关闭 cache
func() {
cacheRepo.Close()
if err := cacheRepo.Close(); err != nil {
loggers.Error("cache close err", zap.Error(err))
} else {
loggers.Info("cache close success")
}
},
// 关闭 gRPC client
func() {
if err := gRPCRepo.Conn().Close(); err != nil {
loggers.Error("gRPC client close err", zap.Error(err))
} else {
loggers.Info("gRPC client close success")
}
},
)
}
pkg/grpclient/client.go
+10 -28
View File
@@ -4,31 +4,23 @@ import (
"context"
"time"
"github.com/xinliangnote/go-gin-api/pkg/trace"
"github.com/pkg/errors"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/keepalive"
"google.golang.org/grpc/resolver"
)
var (
defaultDialTimeout = time.Second * 2
)
type Trace = trace.T
type Option func(*option)
type option struct {
credential credentials.TransportCredentials
keepalive *keepalive.ClientParameters
resolverBuilder resolver.Builder
dialTimeout time.Duration
sign Sign
trace *trace.Trace
grpc *trace.Grpc
credential credentials.TransportCredentials
keepalive *keepalive.ClientParameters
dialTimeout time.Duration
unaryInterceptor grpc.UnaryClientInterceptor
}
// WithCredential setup credential for tls
@@ -52,20 +44,9 @@ func WithDialTimeout(timeout time.Duration) Option {
}
}
// WithSign setup the signature handler
func WithSign(sign Sign) Option {
func WithUnaryInterceptor(unaryInterceptor grpc.UnaryClientInterceptor) Option {
return func(opt *option) {
opt.sign = sign
}
}
// WithTrace setup trace info
func WithTrace(t Trace) Option {
return func(opt *option) {
if t != nil {
opt.trace = t.(*trace.Trace)
opt.grpc = new(trace.Grpc)
}
opt.unaryInterceptor = unaryInterceptor
}
}
@@ -89,12 +70,13 @@ func New(target string, options ...Option) (*grpc.ClientConn, error) {
dialTimeout = opt.dialTimeout
}
clientInterceptor := NewClientInterceptor(opt.sign, opt.trace, opt.grpc)
dialOptions := []grpc.DialOption{
grpc.WithBlock(),
grpc.WithKeepaliveParams(*kacp),
grpc.WithUnaryInterceptor(clientInterceptor.UnaryInterceptor),
}
if opt.unaryInterceptor != nil {
dialOptions = append(dialOptions, grpc.WithUnaryInterceptor(opt.unaryInterceptor))
}
if opt.credential == nil {
pkg/grpclient/interceptor.go
-131
View File
@@ -1,131 +0,0 @@
package grpclient
import (
"context"
"runtime/debug"
"time"
"github.com/xinliangnote/go-gin-api/pkg/time_parse"
"github.com/xinliangnote/go-gin-api/pkg/trace"
"github.com/golang/protobuf/proto"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
)
const (
// Trace ID
TraceID = "trace-id"
)
// ClientInterceptor the client's interceptor
type ClientInterceptor struct {
sign Sign
trace *trace.Trace
grpc *trace.Grpc
}
// NewClientInterceptor create a client interceptor
func NewClientInterceptor(sign Sign, trace *trace.Trace, grpc *trace.Grpc) *ClientInterceptor {
return &ClientInterceptor{
sign: sign,
trace: trace,
grpc: grpc,
}
}
// UnaryInterceptor a interceptor for client unary operations
func (c *ClientInterceptor) UnaryInterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) (err error) {
ts := time.Now()
var invokerErr error
defer func() { // double recover for safety
if p := recover(); p != nil {
debug.PrintStack()
err = status.Errorf(codes.Internal, "got double panic err: %+v,Stack: %s", p, debug.Stack())
return
}
}()
defer func() {
if p := recover(); p != nil {
debug.PrintStack()
err = status.Errorf(codes.Internal, "got panic err: %+v,Stack: %s", p, debug.Stack())
return
}
var mapReq, mapReply map[string]interface{}
mapReq, err = ProtoMessage2Map(req.(proto.Message))
if err != nil {
return
}
mapReply, err = ProtoMessage2Map(reply.(proto.Message))
if err != nil {
return
}
meta, _ := metadata.FromOutgoingContext(ctx)
if c.trace != nil {
c.grpc.Timestamp = time_parse.CSTLayoutString()
c.grpc.Addr = cc.Target()
c.grpc.Method = method
c.grpc.Meta = meta
c.grpc.Request = mapReq
c.grpc.Response = mapReply
c.grpc.CostSeconds = time.Since(ts).Seconds()
if invokerErr != nil {
statusErr, ok := status.FromError(invokerErr)
if ok {
c.grpc.Code = statusErr.Code().String()
c.grpc.Message = statusErr.Message()
}
}
c.trace.AppendGRPC(c.grpc)
}
}()
if c.sign != nil {
var raw string
if req != nil {
if raw, err = ProtoMessage2JSON(req.(proto.Message)); err != nil {
return
}
}
var signature string
if signature, err = c.sign([]byte(raw)); err != nil {
return
}
meta, _ := metadata.FromOutgoingContext(ctx)
if meta == nil {
meta = make(metadata.MD)
}
meta.Set(ProxyAuthorization, signature)
ctx = metadata.NewOutgoingContext(ctx, meta)
}
if c.trace != nil {
meta, _ := metadata.FromOutgoingContext(ctx)
if meta == nil {
meta = make(metadata.MD)
}
meta.Set(TraceID, c.trace.ID())
ctx = metadata.NewOutgoingContext(ctx, meta)
}
invokerErr = invoker(ctx, method, req, reply, cc, opts...)
if invokerErr != nil {
return invokerErr
}
return nil
}
pkg/trace/trace.go
+2
View File
@@ -18,6 +18,8 @@ type T interface {
WithResponse(resp *Response) *Trace
AppendDialog(dialog *Dialog) *Trace
AppendSQL(sql *SQL) *Trace
AppendRedis(redis *Redis) *Trace
AppendGRPC(grpc *Grpc) *Trace
}
// Trace 记录的参数