TrueCloudLab/frostfs-observability
21
0
Fork 10
generated from TrueCloudLab/basic
Code Issues 4 Pull requests Projects Releases Packages Wiki Activity Actions

[#15] tracing: Add events for grpc stream calls
All checks were successful
DCO action / DCO (pull_request) Successful in 1m21s
Details
Tests and linters / Staticcheck (pull_request) Successful in 1m25s
Details
Tests and linters / Tests (pull_request) Successful in 1m40s
Details
Tests and linters / Tests with -race (pull_request) Successful in 1m53s
Details
Tests and linters / Lint (pull_request) Successful in 2m22s
Details

Browse source 
Signed-off-by: Roman Loginov <r.loginov@yadro.com>
This commit is contained in:
Roman Loginov 2024-11-20 11:35:34 +03:00
parent f17779933e
commit f1ed8fd502
3 changed files with 33 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
tracing/grpc/client.go
View file

@ -2,7 +2,10 @@ package grpc
import ( import (
"context" "context"
"fmt"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/metadata" "google.golang.org/grpc/metadata"
) )
@ -10,14 +13,16 @@ import (
type clientStream struct { type clientStream struct {
originalStream grpc.ClientStream originalStream grpc.ClientStream
desc *grpc.StreamDesc desc *grpc.StreamDesc
span trace.Span
finished chan<- error finished chan<- error
done <-chan struct{} 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{ return &clientStream{
originalStream: originalStream, originalStream: originalStream,
desc: desc, desc: desc,
span: span,
finished: finished, finished: finished,
done: done, done: done,
} }
@ -39,6 +44,9 @@ func (cs *clientStream) Trailer() metadata.MD {
} }
func (cs *clientStream) CloseSend() error { func (cs *clientStream) CloseSend() error {
cs.span.AddEvent("closing the send direction of the stream", trace.WithAttributes(
attribute.String("event.name", "client.stream.close.send")),
)
err := cs.originalStream.CloseSend() err := cs.originalStream.CloseSend()
if err != nil { if err != nil {
select { select {
@ -54,6 +62,10 @@ func (cs *clientStream) Context() context.Context {
} }
func (cs *clientStream) SendMsg(m any) error { func (cs *clientStream) SendMsg(m any) error {
cs.span.AddEvent("sending a message", trace.WithAttributes(
attribute.String("event.name", "client.stream.send.msg"),
attribute.String("message.type", fmt.Sprintf("%T", m))),
)
err := cs.originalStream.SendMsg(m) err := cs.originalStream.SendMsg(m)
if err != nil { if err != nil {
select { select {
@ -65,6 +77,10 @@ func (cs *clientStream) SendMsg(m any) error {
} }
func (cs *clientStream) RecvMsg(m any) error { func (cs *clientStream) RecvMsg(m any) error {
cs.span.AddEvent("receiving a message", trace.WithAttributes(
attribute.String("event.name", "client.stream.receive.msg"),
attribute.String("message.type", fmt.Sprintf("%T", m))),
)
err := cs.originalStream.RecvMsg(m) err := cs.originalStream.RecvMsg(m)
if err != nil || !cs.desc.ServerStreams { if err != nil || !cs.desc.ServerStreams {
select { select {

4
tracing/grpc/interceptors.go
View file

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

15
tracing/grpc/server.go
View file

@ -2,7 +2,10 @@ package grpc
import ( import (
"context" "context"
"fmt"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/metadata" "google.golang.org/grpc/metadata"
) )
@ -10,12 +13,14 @@ import (
type serverStream struct { type serverStream struct {
originalStream grpc.ServerStream originalStream grpc.ServerStream
ctx context.Context // nolint:containedctx 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{ return &serverStream{
originalStream: originalStream, originalStream: originalStream,
ctx: ctx, ctx: ctx,
span: span,
} }
} }
@ -36,9 +41,17 @@ func (ss *serverStream) Context() context.Context {
} }
func (ss *serverStream) SendMsg(m any) error { func (ss *serverStream) SendMsg(m any) error {
ss.span.AddEvent("sending a message", trace.WithAttributes(
attribute.String("event.name", "server.stream.send.msg"),
attribute.String("message.type", fmt.Sprintf("%T", m))),
)
return ss.originalStream.SendMsg(m) return ss.originalStream.SendMsg(m)
} }
func (ss *serverStream) RecvMsg(m any) error { func (ss *serverStream) RecvMsg(m any) error {
ss.span.AddEvent("receiving a message", trace.WithAttributes(
attribute.String("event.name", "server.stream.receive.msg"),
attribute.String("message.type", fmt.Sprintf("%T", m))),
)
return ss.originalStream.RecvMsg(m) return ss.originalStream.RecvMsg(m)
} }