*: add drpc client and server metrics

Author: suj-krishnan

drpcconn/conn.go

@@ -111,9 +111,7 @@ func (c *Conn) Close() (err error) { return c.man.Close() }
 
 // Invoke issues the rpc on the transport serializing in, waits for a response, and
 // deserializes it into out. Only one Invoke or Stream may be open at a time.
-func (c *Conn) Invoke(
-	ctx context.Context, rpc string, enc drpc.Encoding, in, out drpc.Message,
-) (err error) {
+func (c *Conn) Invoke(ctx context.Context, rpc string, enc drpc.Encoding, in, out drpc.Message) (err error) {
 	defer func() { err = drpc.ToRPCErr(err) }()
 
 	var metadata []byte
@@ -145,14 +143,7 @@ func (c *Conn) Invoke(
 	return nil
 }
 
-func (c *Conn) doInvoke(
-	stream *drpcstream.Stream,
-	enc drpc.Encoding,
-	rpc string,
-	data []byte,
-	metadata []byte,
-	out drpc.Message,
-) (err error) {
+func (c *Conn) doInvoke(stream *drpcstream.Stream, enc drpc.Encoding, rpc string, data []byte, metadata []byte, out drpc.Message) (err error) {
 	if len(metadata) > 0 {
 		if err := stream.RawWrite(drpcwire.KindInvokeMetadata, metadata); err != nil {
 			return err

drpcmanager/manager_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/zeebo/assert"
 	grpcmetadata "google.golang.org/grpc/metadata"
 	"storj.io/drpc/drpcmetadata"
+
 	"storj.io/drpc/drpctest"
 	"storj.io/drpc/drpcwire"
 )

drpcmetrics/metrics.go

@@ -100,37 +100,34 @@ func StatusCodeLabel(err error) string {
 	return status.Code(err).String()
 }
 
-// WithStatusCode returns a copy of labels with the error_code label added.
-func WithStatusCode(labels map[string]string, err error) map[string]string {
-	out := make(map[string]string, len(labels)+1)
-	for k, v := range labels {
-		out[k] = v
-	}
+// WithStatusCode returns a label map with the RPC method name and status code.
+func WithStatusCode(rpc string, err error) map[string]string {
 	statusCode := drpc.ToRPCErr(err)
-	out[LabelStatusCode] = StatusCodeLabel(statusCode)
-	return out
+	labels := map[string]string{
+		LabelRPCMethodName: rpc,
+		LabelStatusCode: StatusCodeLabel(statusCode),
+	}
+	return labels
 }
 
 // WithState returns a copy of labels with the state label added.
-func WithState(labels map[string]string, state string) map[string]string {
-	out := make(map[string]string, len(labels)+1)
-	for k, v := range labels {
-		out[k] = v
+func WithState(rpc string, state string) map[string]string {
+	labels := map[string]string{
+		LabelRPCMethodName: rpc,
+		LabelState: state,
 	}
-	out[LabelState] = state
-	return out
+	return labels
 }
 
 // WithStateAndStatusCode returns a copy of labels with the state and status code labels added.
-func WithStateAndStatusCode(labels map[string]string,
+func WithStateAndStatusCode(rpc string,
 	state string, err error) map[string]string {
-	out := make(map[string]string, len(labels)+2)
-	for k, v := range labels {
-		out[k] = v
+	labels := map[string]string{
+		LabelRPCMethodName: rpc,
+		LabelState: state,
+		LabelStatusCode: StatusCodeLabel(drpc.ToRPCErr(err)),
 	}
-	out[LabelState] = state
-	out[LabelStatusCode] = StatusCodeLabel(drpc.ToRPCErr(err))
-	return out
+	return labels
 }
 
 // ClientMetrics holds optional metrics that the client connection will

drpcserver/server.go

@@ -32,6 +32,10 @@ type Options struct {
 	// handling individual clients. It is not called if nil.
 	Log func(error)
 
+	// CollectStats controls whether the server should collect stats on the
+	// rpcs it serves.
+	CollectStats bool
+
 	// TLSConfig, if non-nil, is used to wrap the listener with tls.NewListener
 	// in Serve(). The TLS handshake is performed explicitly in ServeOne before
 	// processing requests.
@@ -43,10 +47,6 @@ type Options struct {
 	// restrictions. If it returns a non-nil error the connection is rejected.
 	TLSCipherRestrict func(conn net.Conn) error
 
-	// CollectStats controls whether the server should collect stats on the
-	// rpcs it serves.
-	CollectStats bool
-
 	// Metrics holds optional metrics the server will populate. If nil, no
 	// metrics are recorded.
 	Metrics *ServerMetrics
@@ -86,32 +86,27 @@ func (m *ServerMetrics) wrapTransport(tr drpc.Transport) drpc.Transport {
 	return &drpcmetrics.MeteredTransport{Transport: tr, BytesSent: m.BytesSent, BytesRecv: m.BytesRecv}
 }
 
-// rpcLabels builds a label map for the given RPC string.
-func (s *Server) rpcLabels(rpc string) map[string]string {
-	return map[string]string{
-		drpcmetrics.LabelRPCMethodName: rpc,
-	}
-}
-
 // recordStreamStart records stream start metrics.
-func (s *Server) recordStreamStart(labels map[string]string) {
+func (s *Server) recordStreamStart(rpc string) {
 	sm := s.opts.Metrics
 	if sm != nil && sm.Stream != nil && sm.Stream.StreamsTotal != nil {
-		sm.Stream.StreamsTotal.Inc(drpcmetrics.WithState(labels,
+		sm.Stream.StreamsTotal.Inc(drpcmetrics.WithState(rpc,
 			drpcmetrics.StateStarted), 1)
 	}
 }
 
 // recordStreamEnd records stream end metrics.
-func (s *Server) recordStreamEnd(labels map[string]string, err error, duration time.Duration) {
+func (s *Server) recordStreamEnd(rpc string, err error,
+	duration time.Duration) {
 	sm := s.opts.Metrics
 	if sm != nil && sm.Stream != nil {
 		if sm.Stream.StreamsTotal != nil {
-			sm.Stream.StreamsTotal.Inc(drpcmetrics.WithStateAndStatusCode(labels,
+			sm.Stream.StreamsTotal.Inc(drpcmetrics.WithStateAndStatusCode(rpc,
 				drpcmetrics.StateCompleted, err), 1)
 		}
 		if sm.Stream.StreamDuration != nil {
-			sm.Stream.StreamDuration.Observe(drpcmetrics.WithStatusCode(labels, err),
+			sm.Stream.StreamDuration.Observe(drpcmetrics.WithStatusCode(rpc,
+				err),
 				duration.Seconds())
 		}
 	}
@@ -293,26 +288,24 @@ func (s *Server) Serve(ctx context.Context, lis net.Listener) (err error) {
 
 // handleRPC handles the rpc that has been requested by the stream.
 func (s *Server) handleRPC(stream *drpcstream.Stream, rpc string) (err error) {
-	var labels map[string]string
 	var start time.Time
 	if s.opts.Metrics != nil {
-		labels = s.rpcLabels(rpc)
-		s.recordStreamStart(labels)
+		s.recordStreamStart(rpc)
 		start = time.Now()
 	}
 
 	handlerErr := s.handler.HandleRPC(stream, rpc)
 
 	if handlerErr != nil {
 		sendErr := errs.Wrap(stream.SendError(handlerErr))
-		s.recordStreamEnd(labels, handlerErr, time.Since(start))
+		s.recordStreamEnd(rpc, handlerErr, time.Since(start))
 		return sendErr
 	}
 
 	closeErr := errs.Wrap(stream.CloseSend())
 	// Record success: the handler returned nil, so the RPC itself
 	// succeeded. A CloseSend failure is a transport-level issue and
 	// should not be attributed as the RPC's error code.
-	s.recordStreamEnd(labels, nil, time.Since(start))
+	s.recordStreamEnd(rpc, nil, time.Since(start))
 	return closeErr
 }