[#20] metrics: Add grpc msg send metrics

Streaming RPC has two main metrics: send message and receive message.
But the first one was missed.

Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>

CODEOWNERS

@@ -0,0 +1,3 @@
+.*          @TrueCloudLab/storage-core-committers @TrueCloudLab/storage-core-developers
+.forgejo/.* @potyarkin
+Makefile    @potyarkin

metrics/grpc/client.go

@@ -15,6 +15,9 @@ var clientMetrics = grpcprom.NewClientMetrics(
 	grpcprom.WithClientStreamRecvHistogram(
 		grpcprom.WithHistogramBuckets(prometheus.DefBuckets),
 	),
+	grpcprom.WithClientStreamSendHistogram(
+		grpcprom.WithHistogramBuckets(prometheus.DefBuckets),
+	),
 )
 
 func init() {
@@ -57,6 +60,12 @@ func init() {
 			Help:           "Histogram of response latency (seconds) of the gRPC single message receive.",
 			VariableLabels: []string{"grpc_type", "grpc_service", "grpc_method"},
 		},
+		{
+			Name:           "grpc_client_msg_send_handling_seconds",
+			Type:           dto.MetricType_HISTOGRAM.String(),
+			Help:           "Histogram of response latency (seconds) of the gRPC single message send.",
+			VariableLabels: []string{"grpc_type", "grpc_service", "grpc_method"},
+		},
 	}
 	metrics.MustRegister(clientMetrics, descs...)
 }

tracing/config.go

@@ -3,6 +3,7 @@ package tracing
 import (
 	"crypto/x509"
 	"fmt"
+	"maps"
 )
 
 // Exporter is type of tracing target.
@@ -33,6 +34,9 @@ type Config struct {
 	// Version is version of service instance.
 	// Optional.
 	Version string
+	// Attributes is KV list of attributes.
+	// Optional.
+	Attributes map[string]string
 }
 
 func (c *Config) validate() error {
@@ -81,5 +85,6 @@ func (c *Config) hasChange(other *Config) bool {
 func (c *Config) serviceInfoEqual(other *Config) bool {
 	return c.Service == other.Service &&
 		c.InstanceID == other.InstanceID &&
-		c.Version == other.Version
+		c.Version == other.Version &&
+		maps.Equal(c.Attributes, other.Attributes)
 }

tracing/grpc/client.go

@@ -2,7 +2,10 @@ package grpc
 
 import (
 	"context"
+	"fmt"
 
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/metadata"
 )
@@ -10,14 +13,16 @@ import (
 type clientStream struct {
 	originalStream grpc.ClientStream
 	desc           *grpc.StreamDesc
+	span           trace.Span
 	finished       chan<- error
 	done           <-chan struct{}
 }
 
-func newgRPCClientStream(originalStream grpc.ClientStream, desc *grpc.StreamDesc, finished chan<- error, done <-chan struct{}) grpc.ClientStream {
+func newgRPCClientStream(originalStream grpc.ClientStream, desc *grpc.StreamDesc, span trace.Span, finished chan<- error, done <-chan struct{}) grpc.ClientStream {
 	return &clientStream{
 		originalStream: originalStream,
 		desc:           desc,
+		span:           span,
 		finished:       finished,
 		done:           done,
 	}
@@ -39,6 +44,7 @@ func (cs *clientStream) Trailer() metadata.MD {
 }
 
 func (cs *clientStream) CloseSend() error {
+	cs.span.AddEvent("client.stream.close.send.start")
 	err := cs.originalStream.CloseSend()
 	if err != nil {
 		select {
@@ -46,6 +52,7 @@ func (cs *clientStream) CloseSend() error {
 		case cs.finished <- err:
 		}
 	}
+	cs.span.AddEvent("client.stream.close.send.finish")
 	return err
 }
 
@@ -54,6 +61,9 @@ func (cs *clientStream) Context() context.Context {
 }
 
 func (cs *clientStream) SendMsg(m any) error {
+	cs.span.AddEvent("client.stream.send.msg.start", trace.WithAttributes(
+		attribute.String("message.type", fmt.Sprintf("%T", m))),
+	)
 	err := cs.originalStream.SendMsg(m)
 	if err != nil {
 		select {
@@ -61,10 +71,16 @@ func (cs *clientStream) SendMsg(m any) error {
 		case cs.finished <- err:
 		}
 	}
+	cs.span.AddEvent("client.stream.send.msg.finish", trace.WithAttributes(
+		attribute.String("message.type", fmt.Sprintf("%T", m))),
+	)
 	return err
 }
 
 func (cs *clientStream) RecvMsg(m any) error {
+	cs.span.AddEvent("client.stream.receive.msg.start", trace.WithAttributes(
+		attribute.String("message.type", fmt.Sprintf("%T", m))),
+	)
 	err := cs.originalStream.RecvMsg(m)
 	if err != nil || !cs.desc.ServerStreams {
 		select {
@@ -72,5 +88,8 @@ func (cs *clientStream) RecvMsg(m any) error {
 		case cs.finished <- err:
 		}
 	}
+	cs.span.AddEvent("client.stream.receive.msg.finish", trace.WithAttributes(
+		attribute.String("message.type", fmt.Sprintf("%T", m))),
+	)
 	return err
 }

tracing/grpc/interceptors.go

@@ -50,7 +50,7 @@ func NewStreamClientInterceptor() grpc.StreamClientInterceptor {
 
 		finished := make(chan error)
 		done := make(chan struct{})
-		strWrp := newgRPCClientStream(str, desc, finished, done)
+		strWrp := newgRPCClientStream(str, desc, span, finished, done)
 
 		go func() {
 			defer close(done)
@@ -107,7 +107,7 @@ func NewStreamServerInterceptor() grpc.StreamServerInterceptor {
 			trace.WithSpanKind(trace.SpanKindServer))
 		defer span.End()
 
-		err := handler(srv, newgRPCServerStream(ctx, ss))
+		err := handler(srv, newgRPCServerStream(ctx, ss, span))
 
 		setGRPCSpanStatus(span, err)
 		return err

tracing/grpc/server.go

@@ -2,7 +2,10 @@ package grpc
 
 import (
 	"context"
+	"fmt"
 
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/metadata"
 )
@@ -10,12 +13,14 @@ import (
 type serverStream struct {
 	originalStream grpc.ServerStream
 	ctx            context.Context // nolint:containedctx
+	span           trace.Span
 }
 
-func newgRPCServerStream(ctx context.Context, originalStream grpc.ServerStream) grpc.ServerStream {
+func newgRPCServerStream(ctx context.Context, originalStream grpc.ServerStream, span trace.Span) grpc.ServerStream {
 	return &serverStream{
 		originalStream: originalStream,
 		ctx:            ctx,
+		span:           span,
 	}
 }
 
@@ -36,9 +41,23 @@ func (ss *serverStream) Context() context.Context {
 }
 
 func (ss *serverStream) SendMsg(m any) error {
-	return ss.originalStream.SendMsg(m)
+	ss.span.AddEvent("server.stream.send.msg.start", trace.WithAttributes(
+		attribute.String("message.type", fmt.Sprintf("%T", m))),
+	)
+	err := ss.originalStream.SendMsg(m)
+	ss.span.AddEvent("server.stream.send.msg.finish", trace.WithAttributes(
+		attribute.String("message.type", fmt.Sprintf("%T", m))),
+	)
+	return err
 }
 
 func (ss *serverStream) RecvMsg(m any) error {
-	return ss.originalStream.RecvMsg(m)
+	ss.span.AddEvent("server.stream.receive.msg.start", trace.WithAttributes(
+		attribute.String("message.type", fmt.Sprintf("%T", m))),
+	)
+	err := ss.originalStream.RecvMsg(m)
+	ss.span.AddEvent("server.stream.receive.msg.finish", trace.WithAttributes(
+		attribute.String("message.type", fmt.Sprintf("%T", m))),
+	)
+	return err
 }

tracing/setup.go

@@ -162,6 +162,9 @@ func newResource(cfg *Config) *resource.Resource {
 	if len(cfg.InstanceID) > 0 {
 		attrs = append(attrs, semconv.ServiceInstanceID(cfg.InstanceID))
 	}
+	for k, v := range cfg.Attributes {
+		attrs = append(attrs, attribute.String(k, v))
+	}
 	return resource.NewWithAttributes(
 		semconv.SchemaURL,
 		attrs...,