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
potyarkin:feature/add-codeowners
potyarkin:fix/license
potyarkin:empty
..
compare: potyarkin:fix/license
Branches Tags
potyarkin:feature/add-codeowners
potyarkin:fix/license
potyarkin:empty
TrueCloudLab:master
TrueCloudLab:feat/tracing_attributes

No commits in common. "master" and "fix/license" have entirely different histories.
master ... fix/licens

6 changed files with 7 additions and 56 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

7
tracing/config.go
View file

@ -3,7 +3,6 @@ package tracing
import (
"crypto/x509"
"fmt"
"maps"
)
// Exporter is type of tracing target.
@ -34,9 +33,6 @@ 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 {
@ -85,6 +81,5 @@ 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 &&
maps.Equal(c.Attributes, other.Attributes)
c.Version == other.Version
}

21
tracing/grpc/client.go
View file

@ -2,10 +2,7 @@ package grpc
import (
"context"
"fmt"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
)
@ -13,16 +10,14 @@ 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, 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{
originalStream: originalStream,
desc: desc,
span: span,
finished: finished,
done: done,
}
@ -44,7 +39,6 @@ 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 {
@ -52,7 +46,6 @@ func (cs *clientStream) CloseSend() error {
case cs.finished <- err:
}
}
cs.span.AddEvent("client.stream.close.send.finish")
return err
}
@ -61,9 +54,6 @@ 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 {
@ -71,16 +61,10 @@ 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 {
@ -88,8 +72,5 @@ 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
}

4
tracing/grpc/interceptors.go
View file

@ -50,7 +50,7 @@ func NewStreamClientInterceptor() grpc.StreamClientInterceptor {
finished := make(chan error)
done := make(chan struct{})
strWrp := newgRPCClientStream(str, desc, span, finished, done)
strWrp := newgRPCClientStream(str, desc, 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, span))
err := handler(srv, newgRPCServerStream(ctx, ss))
setGRPCSpanStatus(span, err)
return err

25
tracing/grpc/server.go
View file

@ -2,10 +2,7 @@ package grpc
import (
"context"
"fmt"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
)
@ -13,14 +10,12 @@ import (
type serverStream struct {
originalStream grpc.ServerStream
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{
originalStream: originalStream,
ctx: ctx,
span: span,
}
}
@ -41,23 +36,9 @@ func (ss *serverStream) Context() context.Context {
}
func (ss *serverStream) SendMsg(m any) error {
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
return ss.originalStream.SendMsg(m)
}
func (ss *serverStream) RecvMsg(m any) error {
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
return ss.originalStream.RecvMsg(m)
}

3
tracing/setup.go
View file

@ -162,9 +162,6 @@ 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...,