-
Notifications
You must be signed in to change notification settings - Fork 573
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
otelgrpc: Implement grpc.StatsHandler for trace instrumentation (#3002)
- Loading branch information
1 parent
f44f5ad
commit bf86b26
Showing
4 changed files
with
814 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,45 @@ | ||
// Copyright The OpenTelemetry Authors | ||
// | ||
// Licensed under the Apache License, Version 2.0 (the "License"); | ||
// you may not use this file except in compliance with the License. | ||
// You may obtain a copy of the License at | ||
// | ||
// http://www.apache.org/licenses/LICENSE-2.0 | ||
// | ||
// Unless required by applicable law or agreed to in writing, software | ||
// distributed under the License is distributed on an "AS IS" BASIS, | ||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
// See the License for the specific language governing permissions and | ||
// limitations under the License. | ||
|
||
/* | ||
Package otelgrpc is the instrumentation library for [google.golang.org/grpc] | ||
For now you can instrument your program which use [google.golang.org/grpc] in two ways: | ||
- by [grpc.UnaryClientInterceptor], [grpc.UnaryServerInterceptor], [grpc.StreamClientInterceptor], [grpc.StreamServerInterceptor] | ||
- by [stats.Handler] | ||
Notice: Do not use both interceptors and [stats.Handler] at the same time! If so, you will get duplicated spans and the parent/child relationships between spans will also be broken. | ||
We strongly still recommand you to use [stats.Handler], mainly for two reasons: | ||
Functional advantages: [stats.Handler] has more information for user to build more flexible and granular metric, for example | ||
- multiple different types of represent "data length": In [stats.InPayload], there exists "Length", "CompressedLength", "WireLength" to denote the size of uncompressed, compressed payload data, with or without framing data. But in interceptors, we can only got uncompressed data, and this feature is also removed due to performance problem. | ||
- more accurate timestamp: [stats.InPayload]'s "RecvTime" and [stats.OutPayload]'s "SentTime" records more accurate timestamp that server got and sent the message, the timestamp recorded by interceptors depends on the location of this interceptors in the total interceptor chain. | ||
- some other use cases: for example, catch failure of decoding message. | ||
Performance advantages: If too many interceptors are registered in a service, the interceptor chain can become too long, which increases the latency and processing time of the entire RPC call. | ||
[stats.Handler]: https://pkg.go.dev/google.golang.org/grpc/stats#Handler | ||
[grpc.UnaryClientInterceptor]: https://pkg.go.dev/google.golang.org/grpc#UnaryClientInterceptor | ||
[grpc.UnaryServerInterceptor]: https://pkg.go.dev/google.golang.org/grpc#UnaryServerInterceptor | ||
[grpc.StreamClientInterceptor]: https://pkg.go.dev/google.golang.org/grpc#StreamClientInterceptor | ||
[grpc.StreamServerInterceptor]: https://pkg.go.dev/google.golang.org/grpc#StreamServerInterceptor | ||
[stats.OutPayload]: https://pkg.go.dev/google.golang.org/grpc/stats#OutPayload | ||
[stats.InPayload]: https://pkg.go.dev/google.golang.org/grpc/stats#InPayload | ||
*/ | ||
package otelgrpc // import "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" |
187 changes: 187 additions & 0 deletions
187
instrumentation/google.golang.org/grpc/otelgrpc/stats_handler.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,187 @@ | ||
// Copyright The OpenTelemetry Authors | ||
// | ||
// Licensed under the Apache License, Version 2.0 (the "License"); | ||
// you may not use this file except in compliance with the License. | ||
// You may obtain a copy of the License at | ||
// | ||
// http://www.apache.org/licenses/LICENSE-2.0 | ||
// | ||
// Unless required by applicable law or agreed to in writing, software | ||
// distributed under the License is distributed on an "AS IS" BASIS, | ||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
// See the License for the specific language governing permissions and | ||
// limitations under the License. | ||
|
||
package otelgrpc // import "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" | ||
|
||
import ( | ||
"context" | ||
"sync/atomic" | ||
|
||
grpc_codes "google.golang.org/grpc/codes" | ||
"google.golang.org/grpc/stats" | ||
"google.golang.org/grpc/status" | ||
|
||
"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc/internal" | ||
"go.opentelemetry.io/otel/codes" | ||
semconv "go.opentelemetry.io/otel/semconv/v1.17.0" | ||
"go.opentelemetry.io/otel/trace" | ||
) | ||
|
||
type gRPCContextKey struct{} | ||
|
||
type gRPCContext struct { | ||
messagesReceived int64 | ||
messagesSent int64 | ||
} | ||
|
||
// NewServerHandler creates a stats.Handler for gRPC server. | ||
func NewServerHandler(opts ...Option) stats.Handler { | ||
h := &serverHandler{ | ||
config: newConfig(opts), | ||
} | ||
|
||
h.tracer = h.config.TracerProvider.Tracer( | ||
instrumentationName, | ||
trace.WithInstrumentationVersion(SemVersion()), | ||
) | ||
return h | ||
} | ||
|
||
type serverHandler struct { | ||
*config | ||
tracer trace.Tracer | ||
} | ||
|
||
// TagRPC can attach some information to the given context. | ||
func (h *serverHandler) TagRPC(ctx context.Context, info *stats.RPCTagInfo) context.Context { | ||
ctx = extract(ctx, h.config.Propagators) | ||
|
||
name, attrs := internal.ParseFullMethod(info.FullMethodName) | ||
attrs = append(attrs, RPCSystemGRPC) | ||
ctx, _ = h.tracer.Start( | ||
trace.ContextWithRemoteSpanContext(ctx, trace.SpanContextFromContext(ctx)), | ||
name, | ||
trace.WithSpanKind(trace.SpanKindServer), | ||
trace.WithAttributes(attrs...), | ||
) | ||
|
||
gctx := gRPCContext{} | ||
return context.WithValue(ctx, gRPCContextKey{}, &gctx) | ||
} | ||
|
||
// HandleRPC processes the RPC stats. | ||
func (h *serverHandler) HandleRPC(ctx context.Context, rs stats.RPCStats) { | ||
handleRPC(ctx, rs) | ||
} | ||
|
||
// TagConn can attach some information to the given context. | ||
func (h *serverHandler) TagConn(ctx context.Context, info *stats.ConnTagInfo) context.Context { | ||
span := trace.SpanFromContext(ctx) | ||
attrs := peerAttr(peerFromCtx(ctx)) | ||
span.SetAttributes(attrs...) | ||
return ctx | ||
} | ||
|
||
// HandleConn processes the Conn stats. | ||
func (h *serverHandler) HandleConn(ctx context.Context, info stats.ConnStats) { | ||
} | ||
|
||
// NewClientHandler creates a stats.Handler for gRPC client. | ||
func NewClientHandler(opts ...Option) stats.Handler { | ||
h := &clientHandler{ | ||
config: newConfig(opts), | ||
} | ||
|
||
h.tracer = h.config.TracerProvider.Tracer( | ||
instrumentationName, | ||
trace.WithInstrumentationVersion(SemVersion()), | ||
) | ||
|
||
return h | ||
} | ||
|
||
type clientHandler struct { | ||
*config | ||
tracer trace.Tracer | ||
} | ||
|
||
// TagRPC can attach some information to the given context. | ||
func (h *clientHandler) TagRPC(ctx context.Context, info *stats.RPCTagInfo) context.Context { | ||
name, attrs := internal.ParseFullMethod(info.FullMethodName) | ||
attrs = append(attrs, RPCSystemGRPC) | ||
ctx, _ = h.tracer.Start( | ||
ctx, | ||
name, | ||
trace.WithSpanKind(trace.SpanKindClient), | ||
trace.WithAttributes(attrs...), | ||
) | ||
|
||
gctx := gRPCContext{} | ||
|
||
return inject(context.WithValue(ctx, gRPCContextKey{}, &gctx), h.config.Propagators) | ||
} | ||
|
||
// HandleRPC processes the RPC stats. | ||
func (h *clientHandler) HandleRPC(ctx context.Context, rs stats.RPCStats) { | ||
handleRPC(ctx, rs) | ||
} | ||
|
||
// TagConn can attach some information to the given context. | ||
func (h *clientHandler) TagConn(ctx context.Context, cti *stats.ConnTagInfo) context.Context { | ||
span := trace.SpanFromContext(ctx) | ||
attrs := peerAttr(cti.RemoteAddr.String()) | ||
span.SetAttributes(attrs...) | ||
return ctx | ||
} | ||
|
||
// HandleConn processes the Conn stats. | ||
func (h *clientHandler) HandleConn(context.Context, stats.ConnStats) { | ||
// no-op | ||
} | ||
|
||
func handleRPC(ctx context.Context, rs stats.RPCStats) { | ||
span := trace.SpanFromContext(ctx) | ||
gctx, _ := ctx.Value(gRPCContextKey{}).(*gRPCContext) | ||
var messageId int64 | ||
|
||
switch rs := rs.(type) { | ||
case *stats.Begin: | ||
case *stats.InPayload: | ||
if gctx != nil { | ||
messageId = atomic.AddInt64(&gctx.messagesReceived, 1) | ||
} | ||
span.AddEvent("message", | ||
trace.WithAttributes( | ||
semconv.MessageTypeReceived, | ||
semconv.MessageIDKey.Int64(messageId), | ||
semconv.MessageCompressedSizeKey.Int(rs.CompressedLength), | ||
semconv.MessageUncompressedSizeKey.Int(rs.Length), | ||
), | ||
) | ||
case *stats.OutPayload: | ||
if gctx != nil { | ||
messageId = atomic.AddInt64(&gctx.messagesSent, 1) | ||
} | ||
|
||
span.AddEvent("message", | ||
trace.WithAttributes( | ||
semconv.MessageTypeSent, | ||
semconv.MessageIDKey.Int64(messageId), | ||
semconv.MessageCompressedSizeKey.Int(rs.CompressedLength), | ||
semconv.MessageUncompressedSizeKey.Int(rs.Length), | ||
), | ||
) | ||
case *stats.End: | ||
if rs.Error != nil { | ||
s, _ := status.FromError(rs.Error) | ||
span.SetStatus(codes.Error, s.Message()) | ||
span.SetAttributes(statusCodeAttr(s.Code())) | ||
} else { | ||
span.SetAttributes(statusCodeAttr(grpc_codes.OK)) | ||
} | ||
span.End() | ||
default: | ||
return | ||
} | ||
} |
Oops, something went wrong.