feature: gRPC instrument

.gitignore

@@ -25,4 +25,4 @@ go.work
 # test output
 coverage-diff.html
 coverage-diff.xml
-coverage.txt
+coverage*.*

metrics/common/rpc_enum.go

@@ -3,9 +3,10 @@ package common
 const (
 	RPCSDKGin   = "github.com/gin-gonic/gin"
 	RPCSDKResty = "github.com/go-resty/resty/v2"
+	RPCSDKGRPC  = "google.golang.org/grpc"
 )
 
 const (
 	RPCProtocolHTTP = "http"
-	RPCProtocolRPC  = "rpc"
+	RPCProtocolGRPC = "grpc"
 )

metrics/common/rpc_metrics.go

@@ -9,6 +9,9 @@ import (
 const (
 	DefaultRPCReceiveRequestMetricName = "apm_rpc_receive_request_duration_seconds"
 	DefaultRPCSendRequestMetricName    = "apm_rpc_send_request_duration_seconds"
+
+	DefaultGRPCReceiveMessageTotalName = "apm_grpc_receive_message_total"
+	DefaultGRPCSendMessageTotalName    = "apm_grpc_send_message_total"
 )
 
 var (
@@ -20,7 +23,7 @@ var (
 		NativeHistogramMinResetDuration: 5 * time.Minute,
 		NativeHistogramMaxZeroThreshold: 0.05,
 		NativeHistogramMaxBucketNumber:  20,
-	}, []string{"sdk", "request_protocol", "endpoint", "rpc_status", "response_code"})
+	}, []string{"sdk", "request_protocol", "endpoint", "rpc_status_code", "http_status_code"})
 
 	DefaultRPCSendRequestMetric = prometheus.NewHistogramVec(prometheus.HistogramOpts{
 		Name:                            DefaultRPCSendRequestMetricName,
@@ -30,11 +33,12 @@ var (
 		NativeHistogramMinResetDuration: 5 * time.Minute,
 		NativeHistogramMaxZeroThreshold: 0.05,
 		NativeHistogramMaxBucketNumber:  20,
-	}, []string{"sdk", "request_protocol", "endpoint", "rpc_status", "response_code"})
+	}, []string{"sdk", "request_protocol", "endpoint", "rpc_status_code", "http_status_code"})
 )
 
 func init() {
 	prometheus.MustRegister(
+		// common request
 		DefaultRPCReceiveRequestMetric,
 		DefaultRPCSendRequestMetric,
 	)

metrics/gin/gin.go

@@ -36,8 +36,8 @@ func NewMetricsMiddleware() gin.HandlerFunc {
 			"sdk":              common.RPCSDKGin,
 			"request_protocol": common.RPCProtocolHTTP,
 			"endpoint":         endpoint,
-			"rpc_status":       strconv.Itoa(int(grpc.OK)),
-			"response_code":    strconv.Itoa(responseCode),
+			"rpc_status_code":  strconv.Itoa(int(grpc.OK)),
+			"http_status_code": strconv.Itoa(responseCode),
 		}).Observe(latency.Seconds())
 		return
 	}

metrics/grpc/grpc.go

@@ -0,0 +1,220 @@
+package grpc
+
+import (
+	"context"
+	"io"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/quwan-sre/observability-go-contrib/metrics/common"
+)
+
+func NewUnaryServerInterceptor() grpc.UnaryServerInterceptor {
+	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
+		startTime := time.Now()
+
+		defer func() {
+			latency := time.Now().Sub(startTime)
+			code := codes.OK
+
+			if err != nil {
+				s, _ := status.FromError(err)
+				code = s.Code()
+			}
+
+			common.DefaultRPCReceiveRequestMetric.With(prometheus.Labels{
+				"sdk":              common.RPCSDKGRPC,
+				"request_protocol": common.RPCProtocolGRPC,
+				"endpoint":         ParseFullMethod(info.FullMethod),
+				"rpc_status_code":  strconv.Itoa(int(code)),
+				"http_status_code": "0",
+			}).Observe(latency.Seconds())
+		}()
+
+		resp, err = handler(ctx, req)
+
+		return resp, err
+	}
+}
+
+func NewStreamServerInterceptor() grpc.StreamServerInterceptor {
+	return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) (err error) {
+		err = handler(srv, &wrapServerStream{ss, ParseFullMethod(info.FullMethod)})
+		return err
+	}
+}
+
+func NewUnaryClientInterceptor() grpc.UnaryClientInterceptor {
+	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) (err error) {
+		startTime := time.Now()
+
+		defer func() {
+			latency := time.Now().Sub(startTime)
+			code := codes.OK
+
+			if err != nil {
+				s, _ := status.FromError(err)
+				code = s.Code()
+			}
+
+			common.DefaultRPCSendRequestMetric.With(prometheus.Labels{
+				"sdk":              common.RPCSDKGRPC,
+				"request_protocol": common.RPCProtocolGRPC,
+				"endpoint":         ParseFullMethod(method),
+				"rpc_status_code":  strconv.Itoa(int(code)),
+				"http_status_code": "0",
+			}).Observe(latency.Seconds())
+		}()
+
+		err = invoker(ctx, method, req, reply, cc, opts...)
+		return err
+	}
+}
+
+func NewStreamClientInterceptor() grpc.StreamClientInterceptor {
+	return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
+		s, err := streamer(ctx, desc, cc, method, opts...)
+		return &wrapClientStream{
+			ClientStream: s,
+			method:       ParseFullMethod(method),
+		}, err
+	}
+}
+
+// wrapClientStream wraps grpc.ClientStream to record each Sent/Recv of message in histogram.
+type wrapClientStream struct {
+	grpc.ClientStream
+	method string
+}
+
+func (w *wrapClientStream) SendMsg(m interface{}) error {
+	startTime := time.Now()
+	var err error
+
+	defer func() {
+		latency := time.Now().Sub(startTime)
+		code := codes.OK
+
+		if err != nil && err != io.EOF {
+			s, _ := status.FromError(err)
+			code = s.Code()
+		}
+
+		common.DefaultRPCSendRequestMetric.With(prometheus.Labels{
+			"sdk":              common.RPCSDKGRPC,
+			"request_protocol": common.RPCProtocolGRPC,
+			"endpoint":         w.method,
+			"rpc_status_code":  strconv.Itoa(int(code)),
+			"http_status_code": "0",
+		}).Observe(latency.Seconds())
+	}()
+
+	err = w.ClientStream.SendMsg(m)
+	return err
+}
+
+func (w *wrapClientStream) RecvMsg(m interface{}) error {
+	var err error
+	startTime := time.Now()
+
+	defer func() {
+		latency := time.Now().Sub(startTime)
+		code := codes.OK
+
+		if err != nil && err != io.EOF {
+			s, _ := status.FromError(err)
+			code = s.Code()
+		}
+
+		common.DefaultRPCReceiveRequestMetric.With(prometheus.Labels{
+			"sdk":              common.RPCSDKGRPC,
+			"request_protocol": common.RPCProtocolGRPC,
+			"endpoint":         w.method,
+			"rpc_status_code":  strconv.Itoa(int(code)),
+			"http_status_code": "0",
+		}).Observe(latency.Seconds())
+	}()
+
+	err = w.ClientStream.RecvMsg(m)
+	return err
+}
+
+// wrapServerStream wraps grpc.ServerStream to record each Sent/Recv of message in histogram.
+type wrapServerStream struct {
+	grpc.ServerStream
+	method string
+}
+
+func (w *wrapServerStream) SendMsg(m interface{}) error {
+	startTime := time.Now()
+	var err error
+
+	defer func() {
+		latency := time.Now().Sub(startTime)
+		code := codes.OK
+
+		if err != nil && err != io.EOF {
+			s, _ := status.FromError(err)
+			code = s.Code()
+		}
+
+		common.DefaultRPCSendRequestMetric.With(prometheus.Labels{
+			"sdk":              common.RPCSDKGRPC,
+			"request_protocol": common.RPCProtocolGRPC,
+			"endpoint":         w.method,
+			"rpc_status_code":  strconv.Itoa(int(code)),
+			"http_status_code": "0",
+		}).Observe(latency.Seconds())
+	}()
+
+	err = w.ServerStream.SendMsg(m)
+	return err
+}
+
+func (w *wrapServerStream) RecvMsg(m interface{}) error {
+	var err error
+	startTime := time.Now()
+
+	defer func() {
+		latency := time.Now().Sub(startTime)
+		code := codes.OK
+
+		if err != nil && err != io.EOF {
+			s, _ := status.FromError(err)
+			code = s.Code()
+		}
+
+		common.DefaultRPCReceiveRequestMetric.With(prometheus.Labels{
+			"sdk":              common.RPCSDKGRPC,
+			"request_protocol": common.RPCProtocolGRPC,
+			"endpoint":         w.method,
+			"rpc_status_code":  strconv.Itoa(int(code)),
+			"http_status_code": "0",
+		}).Observe(latency.Seconds())
+	}()
+
+	err = w.ServerStream.RecvMsg(m)
+	return err
+}
+
+// ParseFullMethod returns a "service/method" as endpoint following the OpenTelemetry semantic
+// conventions.
+//
+// Parsing is consistent with grpc-go implementation:
+// https://github.com/grpc/grpc-go/blob/v1.57.0/internal/grpcutil/method.go#L26-L39
+func ParseFullMethod(fullMethod string) string {
+	if !strings.HasPrefix(fullMethod, "/") {
+		// Invalid format, does not follow `/package.service/method`.
+		return fullMethod
+	}
+	name := fullMethod[1:]
+
+	// return in service/method format
+	return name
+}

metrics/resty/resty.go

@@ -46,15 +46,15 @@ func NewAfterResponse() func(c *resty.Client, r *resty.Response) error {
 		latency := time.Now().Sub(t)
 		endpoint := ""
 		if req.RawRequest != nil && req.RawRequest.URL != nil {
-			endpoint = req.RawRequest.URL.Path
+			endpoint = req.RawRequest.URL.Host + req.RawRequest.URL.Path
 		}
 
 		common.DefaultRPCSendRequestMetric.With(prometheus.Labels{
 			"sdk":              common.RPCSDKResty,
 			"request_protocol": common.RPCProtocolHTTP,
 			"endpoint":         endpoint,
-			"rpc_status":       strconv.Itoa(int(grpc.OK)),
-			"response_code":    strconv.Itoa(r.StatusCode()),
+			"rpc_status_code":  strconv.Itoa(int(grpc.OK)),
+			"http_status_code": strconv.Itoa(r.StatusCode()),
 		}).Observe(latency.Seconds())
 		return nil
 	}