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

Compare commits

base: TrueCloudLab:master
Branches Tags
TrueCloudLab:master
TrueCloudLab:feat/tracing_attributes
..
compare: TrueCloudLab:feat/tracing_attributes
Branches Tags
TrueCloudLab:master
TrueCloudLab:feat/tracing_attributes

No commits in common. "master" and "feat/tracing_attributes" have entirely different histories.
master ... feat/traci

6 changed files with 9 additions and 59 deletions
Show stats Expand all files Collapse all files

3
CODEOWNERS
View file

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

9
metrics/grpc/client.go
View file

@ -15,9 +15,6 @@ var clientMetrics = grpcprom.NewClientMetrics(
grpcprom.WithClientStreamRecvHistogram( grpcprom.WithClientStreamRecvHistogram(
grpcprom.WithHistogramBuckets(prometheus.DefBuckets), grpcprom.WithHistogramBuckets(prometheus.DefBuckets),
), ),
grpcprom.WithClientStreamSendHistogram(
grpcprom.WithHistogramBuckets(prometheus.DefBuckets),
),
) )
func init() { func init() {
@ -60,12 +57,6 @@ func init() {
Help: "Histogram of response latency (seconds) of the gRPC single message receive.", Help: "Histogram of response latency (seconds) of the gRPC single message receive.",
VariableLabels: []string{"grpc_type", "grpc_service", "grpc_method"}, 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...) metrics.MustRegister(clientMetrics, descs...)
} }

2
tracing/examples/grpc/main.go
View file

@ -82,7 +82,7 @@ func main() {
cc, err := grpc.NewClient(":7000", cc, err := grpc.NewClient(":7000",
grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithTransportCredentials(insecure.NewCredentials()),
grpc.WithChainUnaryInterceptor( grpc.WithChainUnaryInterceptor(
tracing_grpc.NewUnaryClientInterceptor(), tracing_grpc.NewUnaryClientInteceptor(),
verifyClientTraceID, verifyClientTraceID,
), ),
grpc.WithChainStreamInterceptor( grpc.WithChainStreamInterceptor(

21
tracing/grpc/client.go
View file

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

8
tracing/grpc/interceptors.go
View file

@ -15,8 +15,8 @@ import (
"google.golang.org/grpc/status" "google.golang.org/grpc/status"
) )
// NewUnaryClientInterceptor creates new gRPC unary interceptor to save gRPC client traces. // NewUnaryClientInteceptor creates new gRPC unary interceptor to save gRPC client traces.
func NewUnaryClientInterceptor() grpc.UnaryClientInterceptor { func NewUnaryClientInteceptor() grpc.UnaryClientInterceptor {
return func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { return func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
ctx, span := startClientSpan(ctx, cc, method) ctx, span := startClientSpan(ctx, cc, method)
defer span.End() defer span.End()
@ -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, span, finished, done) strWrp := newgRPCClientStream(str, desc, 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, span)) err := handler(srv, newgRPCServerStream(ctx, ss))
setGRPCSpanStatus(span, err) setGRPCSpanStatus(span, err)
return err return err

25
tracing/grpc/server.go
View file

@ -2,10 +2,7 @@ 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"
) )
@ -13,14 +10,12 @@ 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, span trace.Span) grpc.ServerStream { func newgRPCServerStream(ctx context.Context, originalStream grpc.ServerStream) grpc.ServerStream {
return &serverStream{ return &serverStream{
originalStream: originalStream, originalStream: originalStream,
ctx: ctx, ctx: ctx,
span: span,
} }
} }
@ -41,23 +36,9 @@ func (ss *serverStream) Context() context.Context {
} }
func (ss *serverStream) SendMsg(m any) error { func (ss *serverStream) SendMsg(m any) error {
ss.span.AddEvent("server.stream.send.msg.start", trace.WithAttributes( return ss.originalStream.SendMsg(m)
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 { func (ss *serverStream) RecvMsg(m any) error {
ss.span.AddEvent("server.stream.receive.msg.start", trace.WithAttributes( return ss.originalStream.RecvMsg(m)
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
} }