Merge branch 'errcode' of github.com:OpenIMSDK/Open-IM-Server into errcode

 Conflicts:

	internal/api_to_rpc/api.go

	internal/rpc/group/group.go

cmd/open_im_api/main.go

@@ -15,6 +15,7 @@ import (
 	"Open_IM/internal/api/user"
 	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/log"
+	"Open_IM/pkg/common/middleware"
 	"Open_IM/pkg/utils"
 	"flag"
 	"fmt"
@@ -49,6 +50,7 @@ func main() {
 	r := gin.New()
 	r.Use(gin.Recovery())
 	r.Use(utils.CorsHandler())
+	r.Use(middleware.GinParseOperationID)
 	log.Info("load config: ", config.Config)
 	r.GET("/swagger/*any", ginSwagger.WrapHandler(swaggerFiles.Handler))
 	if config.Config.Prometheus.Enable {

internal/api_to_rpc/api.go

@@ -1,24 +1,22 @@
 package common
 
 import (
-	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/trace_log"
 	utils2 "Open_IM/pkg/utils"
-	"context"
 	utils "github.com/OpenIMSDK/open_utils"
 	"github.com/gin-gonic/gin"
+	"google.golang.org/grpc/metadata"
 	"reflect"
 )
 
 func ApiToRpc(c *gin.Context, apiReq, apiResp interface{}, rpcName string, fn interface{}, rpcFuncName string, tokenFunc func(token string, operationID string) (string, error)) {
+	operationID := c.GetHeader("operationID")
 	nCtx := trace_log.NewCtx(c, rpcFuncName)
 	defer trace_log.ShowLog(nCtx)
 	if err := c.BindJSON(apiReq); err != nil {
 		trace_log.WriteErrorResponse(nCtx, "BindJSON", err)
 		return
 	}
-	reqValue := reflect.ValueOf(apiReq).Elem()
-	operationID := reqValue.FieldByName("OperationID").String()
 	trace_log.SetOperationID(nCtx, operationID)
 	trace_log.SetContextInfo(nCtx, "BindJSON", nil, "params", apiReq)
 	etcdConn, err := utils2.GetConn(c, rpcName)
@@ -39,22 +37,13 @@ func ApiToRpc(c *gin.Context, apiReq, apiResp interface{}, rpcName string, fn in
 			return
 		}
 	}
-	if opID := rpcReqPtr.Elem().FieldByName("OperationID"); opID.IsValid() {
-		opID.SetString(operationID)
-		if opU := rpcReqPtr.Elem().FieldByName("OpUserID"); opU.IsValid() {
-			opU.SetString(opUserID)
-		}
-	} else {
-		op := rpcReqPtr.Elem().FieldByName("Operation").Elem()
-		op.FieldByName("OperationID").SetString(operationID)
-		op.FieldByName("OpUserID").SetString(opUserID)
-	}
 	if err := utils.CopyStructFields(rpcReqPtr.Interface(), apiReq); err != nil {
 		trace_log.WriteErrorResponse(nCtx, "CopyStructFields_RpcReq", err)
 		return
 	}
+	md := metadata.Pairs("operationID", operationID, "opUserID", opUserID)
 	respArr := rpc.Call([]reflect.Value{
-		reflect.ValueOf(context.Context(c)), // context.Context
+		reflect.ValueOf(metadata.NewOutgoingContext(c, md)), // context.Context
 		rpcReqPtr, // rpc apiReq
 	}) // respArr => (apiResp, error)
 	if !respArr[1].IsNil() { // rpc err != nil
@@ -64,17 +53,6 @@ func ApiToRpc(c *gin.Context, apiReq, apiResp interface{}, rpcName string, fn in
 	}
 	rpcResp := respArr[0].Elem()
 	trace_log.SetContextInfo(nCtx, rpcFuncName, nil, "rpc req", rpcReqPtr.Interface(), "resp", rpcResp.Interface())
-	commonResp := rpcResp.FieldByName("CommonResp").Elem()
-	errCodeVal := commonResp.FieldByName("ErrCode")
-	errMsgVal := commonResp.FieldByName("ErrMsg").Interface().(string)
-	errCode := errCodeVal.Interface().(int32)
-	if errCode != 0 {
-		trace_log.WriteErrorResponse(nCtx, "RpcErrCode", &constant.ErrInfo{
-			ErrCode: errCode,
-			ErrMsg:  errMsgVal,
-		})
-		return
-	}
 	if apiResp != nil {
 		if err := utils.CopyStructFields(apiResp, rpcResp.Interface()); err != nil {
 			trace_log.WriteErrorResponse(nCtx, "CopyStructFields_RpcResp", err)
@@ -83,3 +61,77 @@ func ApiToRpc(c *gin.Context, apiReq, apiResp interface{}, rpcName string, fn in
 	}
 	trace_log.SetSuccess(nCtx, rpcFuncName, apiResp)
 }
+
+//func ApiToRpc(c *gin.Context, apiReq, apiResp interface{}, rpcName string, fn interface{}, rpcFuncName string, tokenFunc func(token string, operationID string) (string, error)) {
+//	nCtx := trace_log.NewCtx(c, rpcFuncName)
+//	defer trace_log.ShowLog(nCtx)
+//	if err := c.BindJSON(apiReq); err != nil {
+//		trace_log.WriteErrorResponse(nCtx, "BindJSON", err)
+//		return
+//	}
+//	reqValue := reflect.ValueOf(apiReq).Elem()
+//	operationID := reqValue.FieldByName("OperationID").String()
+//	trace_log.SetOperationID(nCtx, operationID)
+//	trace_log.SetContextInfo(nCtx, "BindJSON", nil, "params", apiReq)
+//	etcdConn, err := utils2.GetConn(c, rpcName)
+//	if err != nil {
+//		trace_log.WriteErrorResponse(nCtx, "GetDefaultConn", err)
+//		return
+//	}
+//	rpc := reflect.ValueOf(fn).Call([]reflect.Value{
+//		reflect.ValueOf(etcdConn),
+//	})[0].MethodByName(rpcFuncName) // rpc func
+//	rpcReqPtr := reflect.New(rpc.Type().In(1).Elem()) // *req参数
+//	var opUserID string
+//	if tokenFunc != nil {
+//		var err error
+//		opUserID, err = tokenFunc(c.GetHeader("token"), operationID)
+//		if err != nil {
+//			trace_log.WriteErrorResponse(nCtx, "TokenFunc", err)
+//			return
+//		}
+//	}
+//	if opID := rpcReqPtr.Elem().FieldByName("OperationID"); opID.IsValid() {
+//		opID.SetString(operationID)
+//		if opU := rpcReqPtr.Elem().FieldByName("OpUserID"); opU.IsValid() {
+//			opU.SetString(opUserID)
+//		}
+//	} else {
+//		op := rpcReqPtr.Elem().FieldByName("Operation").Elem()
+//		op.FieldByName("OperationID").SetString(operationID)
+//		op.FieldByName("OpUserID").SetString(opUserID)
+//	}
+//	if err := utils.CopyStructFields(rpcReqPtr.Interface(), apiReq); err != nil {
+//		trace_log.WriteErrorResponse(nCtx, "CopyStructFields_RpcReq", err)
+//		return
+//	}
+//	respArr := rpc.Call([]reflect.Value{
+//		reflect.ValueOf(context.Context(c)), // context.Context
+//		rpcReqPtr,                           // rpc apiReq
+//	}) // respArr => (apiResp, error)
+//	if !respArr[1].IsNil() { // rpc err != nil
+//		err := respArr[1].Interface().(error)
+//		trace_log.WriteErrorResponse(nCtx, rpcFuncName, err, "rpc req", rpcReqPtr.Interface())
+//		return
+//	}
+//	rpcResp := respArr[0].Elem()
+//	trace_log.SetContextInfo(nCtx, rpcFuncName, nil, "rpc req", rpcReqPtr.Interface(), "resp", rpcResp.Interface())
+//	commonResp := rpcResp.FieldByName("CommonResp").Elem()
+//	errCodeVal := commonResp.FieldByName("ErrCode")
+//	errMsgVal := commonResp.FieldByName("ErrMsg").Interface().(string)
+//	errCode := errCodeVal.Interface().(int32)
+//	if errCode != 0 {
+//		trace_log.WriteErrorResponse(nCtx, "RpcErrCode", &constant.ErrInfo{
+//			ErrCode: errCode,
+//			ErrMsg:  errMsgVal,
+//		})
+//		return
+//	}
+//	if apiResp != nil {
+//		if err := utils.CopyStructFields(apiResp, rpcResp.Interface()); err != nil {
+//			trace_log.WriteErrorResponse(nCtx, "CopyStructFields_RpcResp", err)
+//			return
+//		}
+//	}
+//	trace_log.SetSuccess(nCtx, rpcFuncName, apiResp)
+//}

internal/rpc/fault_tolerant/conn.go

@@ -3,6 +3,7 @@ package fault_tolerant
 import (
 	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/log"
+	"Open_IM/pkg/common/middleware"
 	"Open_IM/pkg/utils"
 	"github.com/OpenIMSDK/getcdv3"
 	"google.golang.org/grpc"
@@ -75,7 +76,7 @@ func GetConfigConn(serviceName string, operationID string) *grpc.ClientConn {
 	}
 	target := rpcRegisterIP + ":" + utils.Int32ToString(int32(configPortList[0]))
 	log.Info(operationID, "rpcRegisterIP ", rpcRegisterIP, " port ", configPortList, " grpc target: ", target, " serviceName: ", serviceName)
-	conn, err := grpc.Dial(target, grpc.WithInsecure())
+	conn, err := grpc.Dial(target, grpc.WithInsecure(), grpc.WithUnaryInterceptor(middleware.RpcClientInterceptor))
 	if err != nil {
 		log.Error(operationID, "grpc.Dail failed ", err.Error())
 		return nil

pkg/common/middleware/gin.go

@@ -0,0 +1,43 @@
+package middleware
+
+import (
+	"bytes"
+	"encoding/json"
+	"github.com/gin-gonic/gin"
+	"io/ioutil"
+	"net/http"
+)
+
+func GinParseOperationID(c *gin.Context) {
+	if c.Request.Method == http.MethodPost {
+		operationID := c.Request.Header.Get("operationID")
+		if operationID == "" {
+			body, err := ioutil.ReadAll(c.Request.Body)
+			if err != nil {
+				c.String(400, "read request body error: "+err.Error())
+				c.Abort()
+				return
+			}
+			req := struct {
+				OperationID string `json:"operationID"`
+			}{}
+			if err := json.Unmarshal(body, &req); err != nil {
+				c.String(400, "get operationID error: "+err.Error())
+				c.Abort()
+				return
+			}
+			if req.OperationID == "" {
+				c.String(400, "operationID empty")
+				c.Abort()
+				return
+			}
+			c.Request.Body = ioutil.NopCloser(bytes.NewReader(body))
+			operationID = req.OperationID
+			c.Request.Header.Set("operationID", operationID)
+		}
+		c.Set("operationID", operationID)
+		c.Next()
+		return
+	}
+	c.Next()
+}

pkg/common/middleware/rpc.go

@@ -0,0 +1,78 @@
+package middleware
+
+import (
+	"Open_IM/pkg/common/constant"
+	"Open_IM/pkg/common/log"
+	"Open_IM/pkg/common/trace_log"
+	"Open_IM/pkg/utils"
+	"context"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/wrapperspb"
+	"path"
+	"runtime/debug"
+)
+
+func RpcServerInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			log.NewError("", info.FullMethod, "panic", r, "stack", string(debug.Stack()))
+		}
+	}()
+	funcName := path.Base(info.FullMethod)
+	md, ok := metadata.FromIncomingContext(ctx)
+	if !ok {
+		return nil, status.New(codes.InvalidArgument, "missing metadata").Err()
+	}
+	var operationID string
+	if opts := md.Get("operationID"); len(opts) != 1 || opts[0] == "" {
+		return nil, status.New(codes.InvalidArgument, "operationID error").Err()
+	} else {
+		operationID = opts[0]
+	}
+	var opUserID string
+	if opts := md.Get("opUserID"); len(opts) != 1 {
+		return nil, status.New(codes.InvalidArgument, "opUserID error").Err()
+	} else {
+		opUserID = opts[0]
+	}
+	ctx = trace_log.NewRpcCtx(ctx, funcName, operationID)
+	defer trace_log.ShowLog(ctx)
+	trace_log.SetContextInfo(ctx, funcName, err, "opUserID", opUserID, "rpcReq", req.(interface{ String() string }).String())
+	resp, err = handler(ctx, req)
+	if err != nil {
+		trace_log.SetContextInfo(ctx, funcName, err)
+		errInfo := constant.ToAPIErrWithErr(err)
+		var code codes.Code
+		if errInfo.ErrCode == 0 {
+			code = codes.Unknown
+		} else {
+			code = codes.Code(errInfo.ErrCode)
+		}
+		sta, err := status.New(code, errInfo.ErrMsg).WithDetails(wrapperspb.String(errInfo.DetailErrMsg))
+		if err != nil {
+			return nil, err
+		}
+		return nil, sta.Err()
+	}
+	trace_log.SetContextInfo(ctx, funcName, nil, "rpcResp", resp.(interface{ String() string }).String())
+	return
+}
+
+func RpcClientInterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) (err error) {
+	//if cc == nil {
+	//	return utils.Wrap(constant.ErrRpcConn, "")
+	//}
+	operationID, ok := ctx.Value("operationID").(string)
+	if !ok {
+		return utils.Wrap(constant.ErrArgs, "ctx missing operationID")
+	}
+	opUserID, ok := ctx.Value("opUserID").(string)
+	if !ok {
+		return utils.Wrap(constant.ErrArgs, "ctx missing opUserID")
+	}
+	md := metadata.Pairs("operationID", operationID, "opUserID", opUserID)
+	return invoker(metadata.NewOutgoingContext(ctx, md), method, req, reply, cc, opts...)
+}