zeromicro/go-zero
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix golint issues (#992)

Browse Source
This commit is contained in:
Kevin Wan
2021-09-04 12:16:30 +08:00
committed by GitHub
parent 20f665ede8
commit 9bdadf2381
14 changed files with 68 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
zrpc/internal/clientinterceptors/opentracinginterceptor.go
View File

@@ -13,9 +13,11 @@ import (
"google.golang.org/grpc/status"
)
// OpenTracingInterceptor returns a grpc.UnaryClientInterceptor for opentelemetry.
func OpenTracingInterceptor() grpc.UnaryClientInterceptor {
propagator := otel.GetTextMapPropagator()
return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn,
invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
if !opentelemetry.Enabled() {
return invoker(ctx, method, req, reply, cc, opts...)
}
@@ -24,13 +26,8 @@ func OpenTracingInterceptor() grpc.UnaryClientInterceptor {
metadataCopy := requestMetadata.Copy()
tr := otel.Tracer(opentelemetry.TraceName)
name, attr := opentelemetry.SpanInfo(method, cc.Target())
var span trace.Span
ctx, span = tr.Start(ctx,
name,
trace.WithSpanKind(trace.SpanKindClient),
trace.WithAttributes(attr...),
)
ctx, span := tr.Start(ctx, name, trace.WithSpanKind(trace.SpanKindClient),
trace.WithAttributes(attr...))
defer span.End()
opentelemetry.Inject(ctx, propagator, &metadataCopy)
@@ -50,30 +47,23 @@ func OpenTracingInterceptor() grpc.UnaryClientInterceptor {
}
}
// StreamOpenTracingInterceptor returns a grpc.StreamClientInterceptor for opentelemetry.
func StreamOpenTracingInterceptor() grpc.StreamClientInterceptor {
propagator := otel.GetTextMapPropagator()
return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string,
streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
if !opentelemetry.Enabled() {
return streamer(ctx, desc, cc, method, opts...)
}
requestMetadata, _ := metadata.FromOutgoingContext(ctx)
metadataCopy := requestMetadata.Copy()
tr := otel.Tracer(opentelemetry.TraceName)
name, attr := opentelemetry.SpanInfo(method, cc.Target())
var span trace.Span
ctx, span = tr.Start(
ctx,
name,
trace.WithSpanKind(trace.SpanKindClient),
trace.WithAttributes(attr...),
)
ctx, span := tr.Start(ctx, name, trace.WithSpanKind(trace.SpanKindClient),
trace.WithAttributes(attr...))
opentelemetry.Inject(ctx, propagator, &metadataCopy)
ctx = metadata.NewOutgoingContext(ctx, metadataCopy)
s, err := streamer(ctx, desc, cc, method, opts...)
if err != nil {
grpcStatus, _ := status.FromError(err)
@@ -82,12 +72,11 @@ func StreamOpenTracingInterceptor() grpc.StreamClientInterceptor {
span.End()
return s, err
}
stream := opentelemetry.WrapClientStream(ctx, s, desc)
go func() {
err := <-stream.Finished
if err != nil {
if err := <-stream.Finished; err != nil {
s, _ := status.FromError(err)
span.SetStatus(codes.Error, s.Message())
span.SetAttributes(opentelemetry.StatusCodeAttr(s.Code()))

6
zrpc/internal/resolver/kube/eventhandler.go
View File

@@ -8,12 +8,14 @@ import (
v1 "k8s.io/api/core/v1"
)
// EventHandler is ResourceEventHandler implementation.
type EventHandler struct {
update func([]string)
endpoints map[string]lang.PlaceholderType
lock sync.Mutex
}
// NewEventHandler returns an EventHandler.
func NewEventHandler(update func([]string)) *EventHandler {
return &EventHandler{
update: update,
@@ -21,6 +23,7 @@ func NewEventHandler(update func([]string)) *EventHandler {
}
}
// OnAdd handles the endpoints add events.
func (h *EventHandler) OnAdd(obj interface{}) {
endpoints, ok := obj.(*v1.Endpoints)
if !ok {
@@ -46,6 +49,7 @@ func (h *EventHandler) OnAdd(obj interface{}) {
}
}
// OnDelete handles the endpoints delete events.
func (h *EventHandler) OnDelete(obj interface{}) {
endpoints, ok := obj.(*v1.Endpoints)
if !ok {
@@ -71,6 +75,7 @@ func (h *EventHandler) OnDelete(obj interface{}) {
}
}
// OnUpdate handles the endpoints update events.
func (h *EventHandler) OnUpdate(oldObj, newObj interface{}) {
oldEndpoints, ok := oldObj.(*v1.Endpoints)
if !ok {
@@ -91,6 +96,7 @@ func (h *EventHandler) OnUpdate(oldObj, newObj interface{}) {
h.Update(newEndpoints)
}
// Update updates the endpoints.
func (h *EventHandler) Update(endpoints *v1.Endpoints) {
h.lock.Lock()
defer h.lock.Unlock()

2
zrpc/internal/resolver/kube/targetparser.go
View File

@@ -15,12 +15,14 @@ const (
var emptyService Service
// Service represents a service with namespace, name and port.
type Service struct {
Namespace string
Name string
Port int
}
// ParseTarget parses the resolver.Target.
func ParseTarget(target resolver.Target) (Service, error) {
var service Service
service.Namespace = target.Authority

34
zrpc/internal/serverinterceptors/opentracinginterceptor.go
View File

@@ -14,6 +14,7 @@ import (
"google.golang.org/grpc/status"
)
// UnaryOpenTracingInterceptor returns a grpc.UnaryServerInterceptor for opentelemetry.
func UnaryOpenTracingInterceptor() grpc.UnaryServerInterceptor {
propagator := otel.GetTextMapPropagator()
return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo,
@@ -24,20 +25,12 @@ func UnaryOpenTracingInterceptor() grpc.UnaryServerInterceptor {
requestMetadata, _ := metadata.FromIncomingContext(ctx)
metadataCopy := requestMetadata.Copy()
bags, spanCtx := opentelemetry.Extract(ctx, propagator, &metadataCopy)
ctx = baggage.ContextWithBaggage(ctx, bags)
tr := otel.Tracer(opentelemetry.TraceName)
name, attr := opentelemetry.SpanInfo(info.FullMethod, opentelemetry.PeerFromCtx(ctx))
var span trace.Span
ctx, span = tr.Start(
trace.ContextWithRemoteSpanContext(ctx, spanCtx),
name,
trace.WithSpanKind(trace.SpanKindServer),
trace.WithAttributes(attr...),
)
ctx, span := tr.Start(trace.ContextWithRemoteSpanContext(ctx, spanCtx), name,
trace.WithSpanKind(trace.SpanKindServer), trace.WithAttributes(attr...))
defer span.End()
opentelemetry.MessageReceived.Event(ctx, 1, req)
@@ -57,6 +50,7 @@ func UnaryOpenTracingInterceptor() grpc.UnaryServerInterceptor {
}
}
// StreamOpenTracingInterceptor returns a grpc.StreamServerInterceptor for opentelemetry.
func StreamOpenTracingInterceptor() grpc.StreamServerInterceptor {
propagator := otel.GetTextMapPropagator()
return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
@@ -67,30 +61,22 @@ func StreamOpenTracingInterceptor() grpc.StreamServerInterceptor {
requestMetadata, _ := metadata.FromIncomingContext(ctx)
metadataCopy := requestMetadata.Copy()
bags, spanCtx := opentelemetry.Extract(ctx, propagator, &metadataCopy)
ctx = baggage.ContextWithBaggage(ctx, bags)
tr := otel.Tracer(opentelemetry.TraceName)
name, attr := opentelemetry.SpanInfo(info.FullMethod, opentelemetry.PeerFromCtx(ctx))
ctx, span := tr.Start(
trace.ContextWithRemoteSpanContext(ctx, spanCtx),
name,
trace.WithSpanKind(trace.SpanKindServer),
trace.WithAttributes(attr...),
)
ctx, span := tr.Start(trace.ContextWithRemoteSpanContext(ctx, spanCtx), name,
trace.WithSpanKind(trace.SpanKindServer), trace.WithAttributes(attr...))
defer span.End()
err := handler(srv, opentelemetry.WrapServerStream(ctx, ss))
if err != nil {
if err := handler(srv, opentelemetry.WrapServerStream(ctx, ss)); err != nil {
s, _ := status.FromError(err)
span.SetStatus(codes.Error, s.Message())
span.SetAttributes(opentelemetry.StatusCodeAttr(s.Code()))
} else {
span.SetAttributes(opentelemetry.StatusCodeAttr(gcodes.OK))
return err
}
return err
span.SetAttributes(opentelemetry.StatusCodeAttr(gcodes.OK))
return nil
}
}