smallnest#564 add TRACE flag to trace rpcx flow, which should be used on in test environment

Author: smallnest

client/client.go

@@ -87,6 +87,7 @@ type RPCClient interface {
 	Call(ctx context.Context, servicePath, serviceMethod string, args interface{}, reply interface{}) error
 	SendRaw(ctx context.Context, r *protocol.Message) (map[string]string, []byte, error)
 	Close() error
+	RemoteAddr() string
 
 	RegisterServerMessageChan(ch chan<- *protocol.Message)
 	UnregisterServerMessageChan()
@@ -123,6 +124,11 @@ func NewClient(option Option) *Client {
 	}
 }
 
+// RemoteAddr returns the remote address.
+func (c *Client) RemoteAddr() string {
+	return c.Conn.RemoteAddr().String()
+}
+
 // Option contains all options for creating clients.
 type Option struct {
 	// Group is used to select the services in the same group. Services set group info in their meta.
@@ -248,6 +254,10 @@ func (client *Client) Go(ctx context.Context, servicePath, serviceMethod string,
 		}
 	}
 	call.Done = done
+
+	if share.Trace {
+		log.Debug("client.Go send request for %s.%s, args: %+v in case of client call", servicePath, serviceMethod, args)
+	}
 	client.send(ctx, call)
 	return call
 }
@@ -313,6 +323,14 @@ func (client *Client) Call(ctx context.Context, servicePath, serviceMethod strin
 func (client *Client) call(ctx context.Context, servicePath, serviceMethod string, args interface{}, reply interface{}) error {
 	seq := new(uint64)
 	ctx = context.WithValue(ctx, seqKey{}, seq)
+
+	if share.Trace {
+		log.Debug("client.call for %s.%s, args: %+v in case of client call", servicePath, serviceMethod, args)
+		defer func() {
+			log.Debug("client.call done for %s.%s, args: %+v in case of client call", servicePath, serviceMethod, args)
+		}()
+	}
+
 	Done := client.Go(ctx, servicePath, serviceMethod, args, reply, make(chan *Call, 1)).Done
 
 	var err error
@@ -558,9 +576,15 @@ func (client *Client) send(ctx context.Context, call *Call) {
 		_ = client.Plugins.DoClientBeforeEncode(req)
 	}
 
+	if share.Trace {
+		log.Debug("client.send for %s.%s, args: %+v in case of client call", call.ServicePath, call.ServiceMethod, call.Args)
+	}
 	allData := req.EncodeSlicePointer()
 	_, err = client.Conn.Write(*allData)
 	protocol.PutData(allData)
+	if share.Trace {
+		log.Debug("client.sent for %s.%s, args: %+v in case of client call", call.ServicePath, call.ServiceMethod, call.Args)
+	}
 
 	if err != nil {
 		client.mutex.Lock()
@@ -620,6 +644,10 @@ func (client *Client) input() {
 			client.mutex.Unlock()
 		}
 
+		if share.Trace {
+			log.Debug("client.input received %v", res)
+		}
+
 		switch {
 		case call == nil:
 			if isServerMessage {

client/xclient.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/juju/ratelimit"
 	ex "github.com/smallnest/rpcx/errors"
+	"github.com/smallnest/rpcx/log"
 	"github.com/smallnest/rpcx/protocol"
 	"github.com/smallnest/rpcx/share"
 	"golang.org/x/sync/singleflight"
@@ -411,10 +412,16 @@ func (c *xClient) Go(ctx context.Context, serviceMethod string, args interface{}
 
 	ctx = setServerTimeout(ctx)
 
+	if share.Trace {
+		log.Debug("select a client for %s.%s, args: %+v in case of xclient Go", c.servicePath, serviceMethod, args)
+	}
 	_, client, err := c.selectClient(ctx, c.servicePath, serviceMethod, args)
 	if err != nil {
 		return nil, err
 	}
+	if share.Trace {
+		log.Debug("selected a client %s for %s.%s, args: %+v in case of xclient Go", client.RemoteAddr(), c.servicePath, serviceMethod, args)
+	}
 	return client.Go(ctx, c.servicePath, serviceMethod, args, reply, done), nil
 }
 
@@ -436,6 +443,10 @@ func (c *xClient) Call(ctx context.Context, serviceMethod string, args interface
 	}
 	ctx = setServerTimeout(ctx)
 
+	if share.Trace {
+		log.Debug("select a client for %s.%s, failMode: %v, args: %+v in case of xclient Call", c.servicePath, serviceMethod, c.failMode, args)
+	}
+
 	var err error
 	k, client, err := c.selectClient(ctx, c.servicePath, serviceMethod, args)
 	if err != nil {
@@ -444,6 +455,10 @@ func (c *xClient) Call(ctx context.Context, serviceMethod string, args interface
 		}
 	}
 
+	if share.Trace {
+		log.Debug("selected a client %s for %s.%s, failMode: %v, args: %+v in case of xclient Call", client.RemoteAddr(), c.servicePath, serviceMethod, c.failMode, args)
+	}
+
 	var e error
 	switch c.failMode {
 	case Failtry:
@@ -613,6 +628,10 @@ func (c *xClient) SendRaw(ctx context.Context, r *protocol.Message) (map[string]
 
 	ctx = setServerTimeout(ctx)
 
+	if share.Trace {
+		log.Debug("select a client for %s.%s, failMode: %v, args: %+v in case of xclient SendRaw", r.ServicePath, r.ServiceMethod, c.failMode, r.Payload)
+	}
+
 	var err error
 	k, client, err := c.selectClient(ctx, r.ServicePath, r.ServiceMethod, r.Payload)
 	if err != nil {
@@ -627,6 +646,10 @@ func (c *xClient) SendRaw(ctx context.Context, r *protocol.Message) (map[string]
 		}
 	}
 
+	if share.Trace {
+		log.Debug("selected a client %s for %s.%s, failMode: %v, args: %+v in case of xclient Call", client.RemoteAddr(), r.ServicePath, r.ServiceMethod, c.failMode, r.Payload)
+	}
+
 	var e error
 	switch c.failMode {
 	case Failtry:
@@ -702,11 +725,19 @@ func (c *xClient) wrapCall(ctx context.Context, client RPCClient, serviceMethod
 		return ErrServerUnavailable
 	}
 
+	if share.Trace {
+		log.Debug("call a client for %s.%s, args: %+v in case of xclient wrapCall", c.servicePath, serviceMethod, args)
+	}
+
 	ctx = share.NewContext(ctx)
 	c.Plugins.DoPreCall(ctx, c.servicePath, serviceMethod, args)
 	err := client.Call(ctx, c.servicePath, serviceMethod, args, reply)
 	c.Plugins.DoPostCall(ctx, c.servicePath, serviceMethod, args, reply, err)
 
+	if share.Trace {
+		log.Debug("called a client for %s.%s, args: %+v, err: %v in case of xclient wrapCall", c.servicePath, serviceMethod, args, err)
+	}
+
 	return err
 }
 

server/server.go

@@ -9,19 +9,18 @@ import (
 	"io"
 	"net"
 	"net/http"
+	"os"
+	"os/exec"
+	"os/signal"
 	"reflect"
 	"regexp"
 	"runtime"
 	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
-	"time"
-
-	"os"
-	"os/exec"
-	"os/signal"
 	"syscall"
+	"time"
 
 	"github.com/smallnest/rpcx/log"
 	"github.com/smallnest/rpcx/protocol"
@@ -245,7 +244,6 @@ func (s *Server) ServeListener(network string, ln net.Listener) (err error) {
 // creating a new service goroutine for each.
 // The service goroutines read requests and then call services to reply to them.
 func (s *Server) serveListener(ln net.Listener) error {
-
 	var tempDelay time.Duration
 
 	s.mu.Lock()
@@ -303,6 +301,10 @@ func (s *Server) serveListener(ln net.Listener) error {
 		s.activeConn[conn] = struct{}{}
 		s.mu.Unlock()
 
+		if share.Trace {
+			log.Debug("server accepted an conn%c", conn.RemoteAddr().String())
+		}
+
 		go s.serveConn(conn)
 	}
 }
@@ -333,6 +335,10 @@ func (s *Server) serveConn(conn net.Conn) {
 			buf = buf[:ss]
 			log.Errorf("serving %s panic error: %s, stack:\n %s", conn.RemoteAddr(), err, buf)
 		}
+		if share.Trace {
+			log.Debug("server closed conn: %v", conn.RemoteAddr().String())
+		}
+
 		s.mu.Lock()
 		delete(s.activeConn, conn)
 		s.mu.Unlock()
@@ -390,8 +396,12 @@ func (s *Server) serveConn(conn net.Conn) {
 			conn.SetWriteDeadline(t0.Add(s.writeTimeout))
 		}
 
+		if share.Trace {
+			log.Debug("server received an request %s from conn: %v", req, conn.RemoteAddr().String())
+		}
+
 		ctx = share.WithLocalValue(ctx, StartRequestContextKey, time.Now().UnixNano())
-		var closeConn = false
+		closeConn := false
 		if !req.IsHeartbeat() {
 			err = s.auth(ctx, req)
 			closeConn = err != nil
@@ -446,15 +456,17 @@ func (s *Server) serveConn(conn net.Conn) {
 
 			s.Plugins.DoPreHandleRequest(ctx, req)
 
+			if share.Trace {
+				log.Debug("server handle request %s from conn: %v", req, conn.RemoteAddr().String())
+			}
 			res, err := s.handleRequest(ctx, req)
-
 			if err != nil {
 				log.Warnf("rpcx: failed to handle request: %v", err)
 			}
 
 			s.Plugins.DoPreWriteResponse(ctx, req, res, err)
 			if !req.IsOneway() {
-				if len(resMetadata) > 0 { //copy meta in context to request
+				if len(resMetadata) > 0 { // copy meta in context to request
 					meta := res.Metadata
 					if meta == nil {
 						res.Metadata = resMetadata
@@ -476,6 +488,10 @@ func (s *Server) serveConn(conn net.Conn) {
 			}
 			s.Plugins.DoPostWriteResponse(ctx, req, res, err)
 
+			if share.Trace {
+				log.Debug("server write response %v for an request %s from conn: %v", res, req, conn.RemoteAddr().String())
+			}
+
 			protocol.FreeMsg(req)
 			protocol.FreeMsg(res)
 		}()
@@ -549,21 +565,26 @@ func (s *Server) handleRequest(ctx context.Context, req *protocol.Message) (res
 	res.SetMessageType(protocol.Response)
 	s.serviceMapMu.RLock()
 	service := s.serviceMap[serviceName]
+
+	if share.Trace {
+		log.Debug("server get service %s for an request %s", service, req)
+	}
+
 	s.serviceMapMu.RUnlock()
 	if service == nil {
 		err = errors.New("rpcx: can't find service " + serviceName)
 		return handleError(res, err)
 	}
 	mtype := service.method[methodName]
 	if mtype == nil {
-		if service.function[methodName] != nil { //check raw functions
+		if service.function[methodName] != nil { // check raw functions
 			return s.handleRequestForFunction(ctx, req)
 		}
 		err = errors.New("rpcx: can't find method " + methodName)
 		return handleError(res, err)
 	}
 
-	var argv = argsReplyPools.Get(mtype.ArgType)
+	argv := argsReplyPools.Get(mtype.ArgType)
 
 	codec := share.Codecs[req.SerializeType()]
 	if codec == nil {
@@ -601,7 +622,6 @@ func (s *Server) handleRequest(ctx context.Context, req *protocol.Message) (res
 			argsReplyPools.Put(mtype.ReplyType, replyv)
 			if err != nil {
 				return handleError(res, err)
-
 			}
 			res.Payload = data
 		}
@@ -614,13 +634,16 @@ func (s *Server) handleRequest(ctx context.Context, req *protocol.Message) (res
 		argsReplyPools.Put(mtype.ReplyType, replyv)
 		if err != nil {
 			return handleError(res, err)
-
 		}
 		res.Payload = data
 	} else if replyv != nil {
 		argsReplyPools.Put(mtype.ReplyType, replyv)
 	}
 
+	if share.Trace {
+		log.Debug("server called service %s for an request %s", service, req)
+	}
+
 	return res, nil
 }
 
@@ -644,7 +667,7 @@ func (s *Server) handleRequestForFunction(ctx context.Context, req *protocol.Mes
 		return handleError(res, err)
 	}
 
-	var argv = argsReplyPools.Get(mtype.ArgType)
+	argv := argsReplyPools.Get(mtype.ArgType)
 
 	codec := share.Codecs[req.SerializeType()]
 	if codec == nil {
@@ -677,7 +700,6 @@ func (s *Server) handleRequestForFunction(ctx context.Context, req *protocol.Mes
 		argsReplyPools.Put(mtype.ReplyType, replyv)
 		if err != nil {
 			return handleError(res, err)
-
 		}
 		res.Payload = data
 	} else if replyv != nil {
@@ -841,7 +863,7 @@ func (s *Server) startProcess() (int, error) {
 	var env []string
 	env = append(env, os.Environ()...)
 
-	var originalWD, _ = os.Getwd()
+	originalWD, _ := os.Getwd()
 	allFiles := []*os.File{os.Stdin, os.Stdout, os.Stderr}
 	process, err := os.StartProcess(argv0, os.Args, &os.ProcAttr{
 		Dir:   originalWD,
@@ -876,7 +898,7 @@ var ip4Reg = regexp.MustCompile(`^(([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-
 func validIP4(ipAddress string) bool {
 	ipAddress = strings.Trim(ipAddress, " ")
 	i := strings.LastIndex(ipAddress, ":")
-	ipAddress = ipAddress[:i] //remove port
+	ipAddress = ipAddress[:i] // remove port
 
 	return ip4Reg.MatchString(ipAddress)
 }

share/share.go

@@ -37,6 +37,11 @@ const (
 	StreamServiceName = "_streamservice"
 )
 
+// Trace is a flag to write a trace log or not.
+// You should not enable this flag ofr product environment and enable it only for test.
+// It writes trace log with logger Debug level.
+var Trace bool
+
 // Codecs are codecs supported by rpcx. You can add customized codecs in Codecs.
 var Codecs = map[protocol.SerializeType]codec.Codec{
 	protocol.SerializeNone: &codec.ByteCodec{},