Introduce grpc interceptors for basic metrics

These will surface the following:

- H: grpc_server_unary_request_duraion
- C: grpc_server_unary_requests_received
- C: grpc_server_unary_requests_completed
- H: grpc_server_stream_request_duration
- C: grpc_server_stream_requests_received
- C: grpc_server_stream_requests_completed
- C: grpc_server_stream_messages_sent
- C: grpc_server_stream_messages_received

All metrics include service and method labels/dimensions and metrics
recorded at the end of a request/stream include the status code at
completion.

FAB-12710 #done

Change-Id: I2bd8bbbd9f50d50340659cded25e89d85890b5e7
Signed-off-by: Matthew Sykes <sykesmat@us.ibm.com>

Author: sykesm

common/grpcmetrics/fakes/echo_service.go

@@ -0,0 +1,28 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package grpcmetrics_test
+
+import (
+	"testing"
+
+	"github.com/hyperledger/fabric/common/grpcmetrics/testpb"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+//go:generate protoc --go_out=plugins=grpc:. testpb/echo.proto
+
+func TestGrpcmetrics(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Grpcmetrics Suite")
+}
+
+//go:generate counterfeiter -o fakes/echo_service.go --fake-name EchoServiceServer . echoServiceServer
+
+type echoServiceServer interface {
+	testpb.EchoServiceServer
+}

common/grpcmetrics/grpcmetrics_suite_test.go

@@ -0,0 +1,98 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package grpcmetrics
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"github.com/hyperledger/fabric/common/metrics"
+	"google.golang.org/grpc"
+)
+
+type UnaryMetrics struct {
+	RequestDuration   metrics.Histogram
+	RequestsReceived  metrics.Counter
+	RequestsCompleted metrics.Counter
+}
+
+func UnaryServerInterceptor(um *UnaryMetrics) grpc.UnaryServerInterceptor {
+	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+		service, method := serviceMethod(info.FullMethod)
+		um.RequestsReceived.With("service", service, "method", method).Add(1)
+
+		startTime := time.Now()
+		resp, err := handler(ctx, req)
+		duration := time.Since(startTime)
+
+		um.RequestDuration.With(
+			"service", service, "method", method, "code", grpc.Code(err).String(),
+		).Observe(float64(duration) / float64(time.Second))
+		um.RequestsCompleted.With("service", service, "method", method, "code", grpc.Code(err).String()).Add(1)
+
+		return resp, err
+	}
+}
+
+type StreamMetrics struct {
+	RequestDuration   metrics.Histogram
+	RequestsReceived  metrics.Counter
+	RequestsCompleted metrics.Counter
+	MessagesSent      metrics.Counter
+	MessagesReceived  metrics.Counter
+}
+
+func StreamServerInterceptor(sm *StreamMetrics) grpc.StreamServerInterceptor {
+	return func(svc interface{}, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+		sm := sm
+		service, method := serviceMethod(info.FullMethod)
+		sm.RequestsReceived.With("service", service, "method", method).Add(1)
+
+		wrappedStream := &serverStream{
+			ServerStream:     stream,
+			messagesSent:     sm.MessagesSent.With("service", service, "method", method),
+			messagesReceived: sm.MessagesReceived.With("service", service, "method", method),
+		}
+
+		startTime := time.Now()
+		err := handler(svc, wrappedStream)
+		duration := time.Since(startTime)
+
+		sm.RequestDuration.With(
+			"service", service, "method", method, "code", grpc.Code(err).String(),
+		).Observe(float64(duration) / float64(time.Second))
+		sm.RequestsCompleted.With("service", service, "method", method, "code", grpc.Code(err).String()).Add(1)
+
+		return err
+	}
+}
+
+func serviceMethod(fullMethod string) (service, method string) {
+	normalizedMethod := strings.Replace(fullMethod, ".", "_", -1)
+	parts := strings.SplitN(normalizedMethod[1:], "/", 2)
+	return parts[0], parts[1]
+}
+
+type serverStream struct {
+	grpc.ServerStream
+	messagesSent     metrics.Counter
+	messagesReceived metrics.Counter
+}
+
+func (ss *serverStream) SendMsg(msg interface{}) error {
+	ss.messagesSent.Add(1)
+	return ss.ServerStream.SendMsg(msg)
+}
+
+func (ss *serverStream) RecvMsg(msg interface{}) error {
+	err := ss.ServerStream.RecvMsg(msg)
+	if err == nil {
+		ss.messagesReceived.Add(1)
+	}
+	return err
+}