feat: instrument grpc interceptor with OpenTelemetry

Author: Chief-Rishab

.github/workflows/build.yml

@@ -19,7 +19,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v3
         with:
-          go-version: "1.18"
+          go-version: "1.20"
           cache: true
           check-latest: true
       - name: Get release tag

.github/workflows/release.yml

@@ -19,7 +19,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v3
         with:
-          go-version: '^1.18'
+          go-version: '^1.20'
           cache: true
           check-latest: true
       - name: Login to DockerHub

.github/workflows/test.yml

@@ -14,7 +14,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v3
         with:
-          go-version: '1.18'
+          go-version: '1.20'
           cache: true
       - name: Install dependencies
         run: sudo apt-get install build-essential
@@ -53,7 +53,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v3
         with:
-          go-version: '1.18'
+          go-version: '1.20'
           cache: true
       - name: Install dependencies
         run: sudo apt-get install build-essential
@@ -74,7 +74,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v3
         with:
-          go-version: '1.18'
+          go-version: '1.20'
           cache: true
       - name: Download coverage
         uses: actions/download-artifact@v3

cmd/lint.go

@@ -28,7 +28,7 @@ func LintCmd() *cobra.Command {
 	return &cobra.Command{
 		Use:     "lint [path]",
 		Aliases: []string{"l"},
-		Args:    cobra.MatchAll(cobra.ExactArgs(1)),
+		Args:    cobra.MatchAll(cobra.ExactArgs(1), cobra.OnlyValidArgs),
 		Short:   "Check for issues in recipes",
 		Long: heredoc.Doc(`
 			Check for issues specified recipes.
@@ -193,14 +193,12 @@ func printConfigError(rcp recipe.Recipe, pluginNode recipe.PluginNode, err plugi
 }
 
 // findPluginByName checks plugin by provided name
-func findPluginByName(plugins []recipe.PluginRecipe, name string) (plugin recipe.PluginRecipe, exists bool) {
-	for _, p := range plugins {
+func findPluginByName(pp []recipe.PluginRecipe, name string) (recipe.PluginRecipe, bool) {
+	for _, p := range pp {
 		if p.Name == name {
-			exists = true
-			plugin = p
-			return
+			return p, true
 		}
 	}
 
-	return
+	return recipe.PluginRecipe{}, false
 }

go.mod

@@ -1,6 +1,6 @@
 module github.com/raystack/meteor
 
-go 1.18
+go 1.20
 
 require (
 	cloud.google.com/go/bigquery v1.52.0
@@ -142,7 +142,7 @@ require (
 	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
 	github.com/google/uuid v1.3.0 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.2.5 // indirect
-	github.com/googleapis/gax-go/v2 v2.12.0 // indirect
+	github.com/googleapis/gax-go/v2 v2.12.0
 	github.com/gopherjs/gopherjs v0.0.0-20210503212227-fb464eba2686 // indirect
 	github.com/gorilla/css v1.0.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.1 // indirect

metrics/otelgrpc/otelgrpc.go

@@ -0,0 +1,198 @@
+package otelgrpc
+
+import (
+	"context"
+	"net"
+	"strings"
+	"time"
+
+	"github.com/raystack/meteor/utils"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+	semconv "go.opentelemetry.io/otel/semconv/v1.20.0"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/peer"
+	"google.golang.org/protobuf/proto"
+)
+
+type UnaryParams struct {
+	Start  time.Time
+	Method string
+	Req    any
+	Res    any
+	Err    error
+}
+
+type Monitor struct {
+	duration     metric.Int64Histogram
+	requestSize  metric.Int64Histogram
+	responseSize metric.Int64Histogram
+	attributes   []attribute.KeyValue
+}
+
+func NewOtelGRPCMonitor(hostName string) Monitor {
+	meter := otel.Meter("github.com/raystack/meteor/metrics/otelgrpc")
+
+	duration, err := meter.Int64Histogram("rpc.client.duration", metric.WithUnit("ms"))
+	handleOtelErr(err)
+
+	requestSize, err := meter.Int64Histogram("rpc.client.request.size", metric.WithUnit("By"))
+	handleOtelErr(err)
+
+	responseSize, err := meter.Int64Histogram("rpc.client.response.size", metric.WithUnit("By"))
+	handleOtelErr(err)
+
+	addr, port := ExtractAddress(hostName)
+
+	return Monitor{
+		duration:     duration,
+		requestSize:  requestSize,
+		responseSize: responseSize,
+		attributes: []attribute.KeyValue{
+			semconv.RPCSystemGRPC,
+			attribute.String("network.transport", "tcp"),
+			attribute.String("server.address", addr),
+			attribute.String("server.port", port),
+		},
+	}
+}
+
+func GetProtoSize(p any) int {
+	if p == nil {
+		return 0
+	}
+
+	size := proto.Size(p.(proto.Message))
+	return size
+}
+
+func (m *Monitor) RecordUnary(ctx context.Context, p UnaryParams) {
+	reqSize := GetProtoSize(p.Req)
+	resSize := GetProtoSize(p.Res)
+
+	attrs := make([]attribute.KeyValue, len(m.attributes))
+	copy(attrs, m.attributes)
+	attrs = append(attrs, attribute.String("rpc.grpc.status_text", utils.StatusText(p.Err)))
+	attrs = append(attrs, attribute.String("network.type", netTypeFromCtx(ctx)))
+	attrs = append(attrs, ParseFullMethod(p.Method)...)
+
+	m.duration.Record(ctx,
+		time.Since(p.Start).Milliseconds(),
+		metric.WithAttributes(attrs...))
+
+	m.requestSize.Record(ctx,
+		int64(reqSize),
+		metric.WithAttributes(attrs...))
+
+	m.responseSize.Record(ctx,
+		int64(resSize),
+		metric.WithAttributes(attrs...))
+}
+
+func (m *Monitor) RecordStream(ctx context.Context, start time.Time, method string, err error) {
+	attrs := make([]attribute.KeyValue, len(m.attributes))
+	copy(attrs, m.attributes)
+	attrs = append(attrs, attribute.String("rpc.grpc.status_text", utils.StatusText(err)))
+	attrs = append(attrs, attribute.String("network.type", netTypeFromCtx(ctx)))
+	attrs = append(attrs, ParseFullMethod(method)...)
+
+	m.duration.Record(ctx,
+		time.Since(start).Milliseconds(),
+		metric.WithAttributes(attrs...))
+}
+
+func (m *Monitor) UnaryClientInterceptor() grpc.UnaryClientInterceptor {
+	return func(ctx context.Context,
+		method string,
+		req, reply interface{},
+		cc *grpc.ClientConn,
+		invoker grpc.UnaryInvoker,
+		opts ...grpc.CallOption,
+	) (err error) {
+		defer func(start time.Time) {
+			m.RecordUnary(ctx, UnaryParams{
+				Start: start,
+				Req:   req,
+				Res:   reply,
+				Err:   err,
+			})
+		}(time.Now())
+
+		return invoker(ctx, method, req, reply, cc, opts...)
+	}
+}
+
+func (m *Monitor) StreamClientInterceptor() grpc.StreamClientInterceptor {
+	return func(ctx context.Context,
+		desc *grpc.StreamDesc,
+		cc *grpc.ClientConn,
+		method string,
+		streamer grpc.Streamer,
+		opts ...grpc.CallOption,
+	) (s grpc.ClientStream, err error) {
+		defer func(start time.Time) {
+			m.RecordStream(ctx, start, method, err)
+		}(time.Now())
+
+		return streamer(ctx, desc, cc, method, opts...)
+	}
+}
+
+func (m *Monitor) GetAttributes() []attribute.KeyValue {
+	return m.attributes
+}
+
+func ParseFullMethod(fullMethod string) []attribute.KeyValue {
+	name := strings.TrimLeft(fullMethod, "/")
+	service, method, found := strings.Cut(name, "/")
+	if !found {
+		return nil
+	}
+
+	var attrs []attribute.KeyValue
+	if service != "" {
+		attrs = append(attrs, semconv.RPCService(service))
+	}
+	if method != "" {
+		attrs = append(attrs, semconv.RPCMethod(method))
+	}
+	return attrs
+}
+
+func handleOtelErr(err error) {
+	if err != nil {
+		otel.Handle(err)
+	}
+}
+
+func ExtractAddress(addr string) (host, port string) {
+	host, port, err := net.SplitHostPort(addr)
+	if err != nil {
+		return addr, "80"
+	}
+
+	return host, port
+}
+
+func netTypeFromCtx(ctx context.Context) (ipType string) {
+	ipType = "unknown"
+	p, ok := peer.FromContext(ctx)
+	if !ok {
+		return ipType
+	}
+
+	clientIP, _, err := net.SplitHostPort(p.Addr.String())
+	if err != nil {
+		return ipType
+	}
+
+	ip := net.ParseIP(clientIP)
+	if ip.To4() != nil {
+		ipType = "ipv4"
+	} else if ip.To16() != nil {
+		ipType = "ipv6"
+	}
+
+	return ipType
+}

metrics/otelgrpc/otelgrpc_test.go

@@ -0,0 +1,106 @@
+package otelgrpc_test
+
+import (
+	"context"
+	"errors"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/raystack/meteor/metrics/otelgrpc"
+	pb "github.com/raystack/optimus/protos/raystack/optimus/core/v1beta1"
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/otel/attribute"
+	semconv "go.opentelemetry.io/otel/semconv/v1.20.0"
+)
+
+func Test_otelGRPCMonitor_Record(t *testing.T) {
+	mt := otelgrpc.NewOtelGRPCMonitor("localhost:1001")
+	assert.NotNil(t, mt)
+	initialAttr := mt.GetAttributes()
+
+	uc := mt.UnaryClientInterceptor()
+	assert.NotNil(t, uc)
+	assert.Equal(t, initialAttr, mt.GetAttributes())
+
+	sc := mt.StreamClientInterceptor()
+	assert.NotNil(t, sc)
+	assert.Equal(t, initialAttr, mt.GetAttributes())
+
+	mt.RecordUnary(context.Background(), otelgrpc.UnaryParams{
+		Start:  time.Now(),
+		Method: "/service.raystack.com/MethodName",
+		Req:    nil,
+		Res:    nil,
+		Err:    nil,
+	})
+	assert.Equal(t, initialAttr, mt.GetAttributes())
+
+	mt.RecordUnary(context.Background(), otelgrpc.UnaryParams{
+		Start:  time.Now(),
+		Method: "",
+		Req:    &pb.ListProjectsRequest{},
+		Res:    nil,
+		Err:    nil,
+	})
+	assert.Equal(t, initialAttr, mt.GetAttributes())
+
+	mt.RecordStream(context.Background(), time.Now(), "", nil)
+	assert.Equal(t, initialAttr, mt.GetAttributes())
+
+	mt.RecordStream(context.Background(), time.Now(), "/service.raystack.com/MethodName", errors.New("dummy error"))
+	assert.Equal(t, initialAttr, mt.GetAttributes())
+}
+
+func Test_parseFullMethod(t *testing.T) {
+	type args struct {
+		fullMethod string
+	}
+	tests := []struct {
+		name string
+		args args
+		want []attribute.KeyValue
+	}{
+		{name: "should parse correct method", args: args{
+			fullMethod: "/test.service.name/MethodNameV1",
+		}, want: []attribute.KeyValue{
+			semconv.RPCService("test.service.name"),
+			semconv.RPCMethod("MethodNameV1"),
+		}},
+
+		{name: "should return empty attributes on incorrect method", args: args{
+			fullMethod: "incorrectMethod",
+		}, want: nil},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := otelgrpc.ParseFullMethod(tt.args.fullMethod); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("parseFullMethod() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_getProtoSize(t *testing.T) {
+	req := &pb.ListProjectNamespacesRequest{
+		ProjectName: "asd",
+	}
+
+	if got := otelgrpc.GetProtoSize(req); got != 5 {
+		t.Errorf("getProtoSize() = %v, want %v", got, 5)
+	}
+}
+
+func TestExtractAddress(t *testing.T) {
+	gotHost, gotPort := otelgrpc.ExtractAddress("localhost:1001")
+	assert.Equal(t, "localhost", gotHost)
+	assert.Equal(t, "1001", gotPort)
+
+	gotHost, gotPort = otelgrpc.ExtractAddress("localhost")
+	assert.Equal(t, "localhost", gotHost)
+	assert.Equal(t, "80", gotPort)
+
+	gotHost, gotPort = otelgrpc.ExtractAddress("some.address.golabs.io:15010")
+	assert.Equal(t, "some.address.golabs.io", gotHost)
+	assert.Equal(t, "15010", gotPort)
+}