Skip to content

Commit

Permalink
add config for rpc.server.duration
Browse files Browse the repository at this point in the history
add tests

update change log and split up test

remove accidental dependabot change
  • Loading branch information
chchaffin committed Apr 18, 2023
1 parent f998e3f commit caa8291
Show file tree
Hide file tree
Showing 7 changed files with 200 additions and 14 deletions.
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ Thumbs.db
.tools/
.idea/
.vscode/
vendor/
*.iml
*.so
coverage.*
Expand Down
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
### Fixed

- Prevent taking from reservoir in AWS XRay Remote Sampler when there is zero capacity in `go.opentelemetry.io/contrib/samplers/aws/xray`. (#3684)
- Improved `go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc` to optionally, off by default, use peer information as an attribute for the meter rpc.server.duration (#3536)

## [1.16.0-rc.2/0.41.0-rc.2/0.10.0-rc.2] - 2023-03-23

Expand Down
19 changes: 17 additions & 2 deletions instrumentation/google.golang.org/grpc/otelgrpc/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -43,8 +43,9 @@ type config struct {
TracerProvider trace.TracerProvider
MeterProvider metric.MeterProvider

meter metric.Meter
rpcServerDuration instrument.Int64Histogram
meter metric.Meter
rpcServerDuration instrument.Int64Histogram
includeRPCServerDurationPeerCtx bool
}

// Option applies an option value for a config.
Expand Down Expand Up @@ -127,6 +128,20 @@ func (o meterProviderOption) apply(c *config) {
}
}

type includePeerAddrAsAttribute struct {
use bool
}

// WithPeerAddrAsAttribute returns an Option to use peer details when
// recording with the meter rpc.server.duration.
func WithPeerAddrAsAttribute(peerAddrAsAttribute bool) Option {
return includePeerAddrAsAttribute{use: peerAddrAsAttribute}
}

func (o includePeerAddrAsAttribute) apply(c *config) {
c.includeRPCServerDurationPeerCtx = o.use
}

// WithMeterProvider returns an Option to use the MeterProvider when
// creating a Meter. If this option is not provide the global MeterProvider will be used.
func WithMeterProvider(mp metric.MeterProvider) Option {
Expand Down
12 changes: 10 additions & 2 deletions instrumentation/google.golang.org/grpc/otelgrpc/interceptor.go
Original file line number Diff line number Diff line change
Expand Up @@ -321,12 +321,16 @@ func UnaryServerInterceptor(opts ...Option) grpc.UnaryServerInterceptor {

ctx = extract(ctx, cfg.Propagators)

name, attr := spanInfo(info.FullMethod, peerFromCtx(ctx))
name, nameAttr := internal.ParseFullMethod(info.FullMethod)
peerAttrs := peerAttr(peerFromCtx(ctx))
spanAttrs := append(nameAttr, peerAttrs...)
spanAttrs = append(spanAttrs, RPCSystemGRPC)

ctx, span := tracer.Start(
trace.ContextWithRemoteSpanContext(ctx, trace.SpanContextFromContext(ctx)),
name,
trace.WithSpanKind(trace.SpanKindServer),
trace.WithAttributes(attr...),
trace.WithAttributes(spanAttrs...),
)
defer span.End()

Expand All @@ -335,6 +339,10 @@ func UnaryServerInterceptor(opts ...Option) grpc.UnaryServerInterceptor {
var statusCode grpc_codes.Code
defer func(t time.Time) {
elapsedTime := time.Since(t) / time.Millisecond
attr := append(nameAttr, RPCSystemGRPC)
if cfg.includeRPCServerDurationPeerCtx {
attr = append(attr, peerAttrs...)
}
attr = append(attr, semconv.RPCGRPCStatusCodeKey.Int64(int64(statusCode)))
cfg.rpcServerDuration.Record(ctx, int64(elapsedTime), attr...)
}(time.Now())
Expand Down
102 changes: 102 additions & 0 deletions instrumentation/google.golang.org/grpc/otelgrpc/test/bufconn_mock.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,102 @@
package test

import (
"context"
"fmt"
"google.golang.org/grpc/test/bufconn"
"net"
"time"
)

const (
mockIp = "1.1.1.1"
mockPort = 1234
)

var (
mockAddr, _ = net.ResolveTCPAddr("tcp", fmt.Sprintf("%s:%d", mockIp, mockPort))
)

// bufConnMock wraps a bufconn.Lister for the sake of returning a valid address. This is so we can properly test
// returning net.sock.peer.addr and net.sock.peer.port attributes
type bufConnMock struct {
listener *bufconn.Listener
}

func NewMockBufConn(size int) *bufConnMock {
return &bufConnMock{
listener: bufconn.Listen(size),
}
}

func (b *bufConnMock) Accept() (net.Conn, error) {
conn, err := b.listener.Accept()
if err != nil {
return nil, err
}

return &bufConn{
conn: conn,
}, nil
}

func (b *bufConnMock) Close() error {
return b.listener.Close()
}

func (b *bufConnMock) Addr() net.Addr {
return mockAddr
}

func (b *bufConnMock) Dial() (net.Conn, error) {
// bufConnect's listener Dial implementation just calls
// DialContext under the covers so don't wrap the connection in our mock here
return b.listener.DialContext(context.Background())
}

func (b *bufConnMock) DialContext(ctx context.Context) (net.Conn, error) {
conn, err := b.listener.DialContext(ctx)
if err != nil {
return nil, err
}

return &bufConn{
conn: conn,
}, nil
}

type bufConn struct {
conn net.Conn
}

func (b *bufConn) Read(bytes []byte) (n int, err error) {
return b.conn.Read(bytes)
}

func (b *bufConn) Write(bytes []byte) (n int, err error) {
return b.conn.Write(bytes)
}

func (b *bufConn) Close() error {
return b.conn.Close()
}

func (b *bufConn) LocalAddr() net.Addr {
return mockAddr
}

func (b *bufConn) RemoteAddr() net.Addr {
return mockAddr
}

func (b *bufConn) SetDeadline(t time.Time) error {
return b.conn.SetDeadline(t)
}

func (b *bufConn) SetReadDeadline(t time.Time) error {
return b.conn.SetReadDeadline(t)
}

func (b *bufConn) SetWriteDeadline(t time.Time) error {
return b.conn.SetWriteDeadline(t)
}
63 changes: 55 additions & 8 deletions instrumentation/google.golang.org/grpc/otelgrpc/test/grpc_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@ import (
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/interop"
pb "google.golang.org/grpc/interop/grpc_testing"
"google.golang.org/grpc/test/bufconn"

"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
"go.opentelemetry.io/otel/attribute"
Expand All @@ -40,7 +39,7 @@ import (
const bufSize = 2048

func doCalls(cOpt []grpc.DialOption, sOpt []grpc.ServerOption) error {
l := bufconn.Listen(bufSize)
l := NewMockBufConn(bufSize)
defer l.Close()

s := grpc.NewServer(sOpt...)
Expand All @@ -56,7 +55,7 @@ func doCalls(cOpt []grpc.DialOption, sOpt []grpc.ServerOption) error {
dial := func(context.Context, string) (net.Conn, error) { return l.Dial() }
conn, err := grpc.DialContext(
ctx,
"bufnet",
l.Addr().String(),
append([]grpc.DialOption{
grpc.WithContextDialer(dial),
grpc.WithTransportCredentials(insecure.NewCredentials()),
Expand Down Expand Up @@ -113,14 +112,33 @@ func TestInterceptors(t *testing.T) {

t.Run("UnaryServerSpans", func(t *testing.T) {
checkUnaryServerSpans(t, serverUnarySR.Ended())
checkUnaryServerRecords(t, serverUnaryMetricReader)
checkUnaryServerRecords(t, serverUnaryMetricReader, false)
})

t.Run("StreamServerSpans", func(t *testing.T) {
checkStreamServerSpans(t, serverStreamSR.Ended())
})
}

func TestInterceptorWithPeerAddrAsAttribute(t *testing.T) {
serverUnaryPeerAddrSR := tracetest.NewSpanRecorder()
serverUnaryPeerAddrTP := trace.NewTracerProvider(trace.WithSpanProcessor(serverUnaryPeerAddrSR))
serverUnaryMetricReaderPeerAddr := metric.NewManualReader()
serverUnaryMPPeerAddr := metric.NewMeterProvider(metric.WithReader(serverUnaryMetricReaderPeerAddr))

assert.NoError(t, doCalls(nil,
[]grpc.ServerOption{
grpc.UnaryInterceptor(otelgrpc.UnaryServerInterceptor(otelgrpc.WithTracerProvider(serverUnaryPeerAddrTP), otelgrpc.WithMeterProvider(serverUnaryMPPeerAddr), otelgrpc.WithPeerAddrAsAttribute(true))),
},
))

t.Run("UnaryServerSpansWithPeerAddr", func(t *testing.T) {
checkUnaryServerSpans(t, serverUnaryPeerAddrSR.Ended())
checkUnaryServerRecords(t, serverUnaryMetricReaderPeerAddr, true)
})

}

func checkUnaryClientSpans(t *testing.T, spans []trace.ReadOnlySpan) {
require.Len(t, spans, 2)

Expand Down Expand Up @@ -148,6 +166,8 @@ func checkUnaryClientSpans(t *testing.T, spans []trace.ReadOnlySpan) {
semconv.RPCService("grpc.testing.TestService"),
otelgrpc.RPCSystemGRPC,
otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
semconv.NetSockPeerAddr(mockIp),
semconv.NetSockPeerPort(mockPort),
}, emptySpan.Attributes())

largeSpan := spans[1]
Expand Down Expand Up @@ -176,6 +196,8 @@ func checkUnaryClientSpans(t *testing.T, spans []trace.ReadOnlySpan) {
semconv.RPCService("grpc.testing.TestService"),
otelgrpc.RPCSystemGRPC,
otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
semconv.NetSockPeerAddr(mockIp),
semconv.NetSockPeerPort(mockPort),
}, largeSpan.Attributes())
}

Expand Down Expand Up @@ -222,6 +244,8 @@ func checkStreamClientSpans(t *testing.T, spans []trace.ReadOnlySpan) {
semconv.RPCService("grpc.testing.TestService"),
otelgrpc.RPCSystemGRPC,
otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
semconv.NetSockPeerAddr(mockIp),
semconv.NetSockPeerPort(mockPort),
}, streamInput.Attributes())

streamOutput := spans[1]
Expand Down Expand Up @@ -270,6 +294,8 @@ func checkStreamClientSpans(t *testing.T, spans []trace.ReadOnlySpan) {
semconv.RPCService("grpc.testing.TestService"),
otelgrpc.RPCSystemGRPC,
otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
semconv.NetSockPeerAddr(mockIp),
semconv.NetSockPeerPort(mockPort),
}, streamOutput.Attributes())

pingPong := spans[2]
Expand Down Expand Up @@ -338,6 +364,8 @@ func checkStreamClientSpans(t *testing.T, spans []trace.ReadOnlySpan) {
semconv.RPCService("grpc.testing.TestService"),
otelgrpc.RPCSystemGRPC,
otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
semconv.NetSockPeerAddr(mockIp),
semconv.NetSockPeerPort(mockPort),
}, pingPong.Attributes())
}

Expand Down Expand Up @@ -390,6 +418,8 @@ func checkStreamServerSpans(t *testing.T, spans []trace.ReadOnlySpan) {
semconv.RPCService("grpc.testing.TestService"),
otelgrpc.RPCSystemGRPC,
otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
semconv.NetSockPeerAddr(mockIp),
semconv.NetSockPeerPort(mockPort),
}, streamInput.Attributes())

streamOutput := spans[1]
Expand Down Expand Up @@ -438,6 +468,8 @@ func checkStreamServerSpans(t *testing.T, spans []trace.ReadOnlySpan) {
semconv.RPCService("grpc.testing.TestService"),
otelgrpc.RPCSystemGRPC,
otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
semconv.NetSockPeerAddr(mockIp),
semconv.NetSockPeerPort(mockPort),
}, streamOutput.Attributes())

pingPong := spans[2]
Expand Down Expand Up @@ -506,6 +538,8 @@ func checkStreamServerSpans(t *testing.T, spans []trace.ReadOnlySpan) {
semconv.RPCService("grpc.testing.TestService"),
otelgrpc.RPCSystemGRPC,
otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
semconv.NetSockPeerAddr(mockIp),
semconv.NetSockPeerPort(mockPort),
}, pingPong.Attributes())
}

Expand Down Expand Up @@ -536,6 +570,8 @@ func checkUnaryServerSpans(t *testing.T, spans []trace.ReadOnlySpan) {
semconv.RPCService("grpc.testing.TestService"),
otelgrpc.RPCSystemGRPC,
otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
semconv.NetSockPeerAddr(mockIp),
semconv.NetSockPeerPort(mockPort),
}, emptySpan.Attributes())

largeSpan := spans[1]
Expand Down Expand Up @@ -564,6 +600,8 @@ func checkUnaryServerSpans(t *testing.T, spans []trace.ReadOnlySpan) {
semconv.RPCService("grpc.testing.TestService"),
otelgrpc.RPCSystemGRPC,
otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
semconv.NetSockPeerAddr(mockIp),
semconv.NetSockPeerPort(mockPort),
}, largeSpan.Attributes())
}

Expand All @@ -585,7 +623,7 @@ func assertEvents(t *testing.T, expected, actual []trace.Event) bool {
return !failed
}

func checkUnaryServerRecords(t *testing.T, reader metric.Reader) {
func checkUnaryServerRecords(t *testing.T, reader metric.Reader, peerInfoIncluded bool) {
rm := metricdata.ResourceMetrics{}
err := reader.Collect(context.Background(), &rm)
assert.NoError(t, err)
Expand All @@ -598,12 +636,21 @@ func checkUnaryServerRecords(t *testing.T, reader metric.Reader) {
attr := dpt.Attributes.ToSlice()
method := getRPCMethod(attr)
assert.NotEmpty(t, method)
assert.ElementsMatch(t, []attribute.KeyValue{

attrs := []attribute.KeyValue{
otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
semconv.RPCMethod(method),
semconv.RPCService("grpc.testing.TestService"),
otelgrpc.RPCSystemGRPC,
otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
}, attr)
}

if peerInfoIncluded {
attrs = append(attrs, []attribute.KeyValue{
semconv.NetSockPeerPort(mockPort),
semconv.NetSockPeerAddr(mockIp),
}...)
}
assert.ElementsMatch(t, attrs, attr)
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -694,10 +694,22 @@ func assertServerSpan(t *testing.T, wantSpanCode codes.Code, wantSpanStatusDescr
}

// TestUnaryServerInterceptor tests the server interceptor for unary RPCs.
func TestUnaryServerInterceptor(t *testing.T) {
func TestUnaryServerInterceptorWithPeerAddrAsAttribute(t *testing.T) {
sr := tracetest.NewSpanRecorder()
tp := trace.NewTracerProvider(trace.WithSpanProcessor(sr))
usi := otelgrpc.UnaryServerInterceptor(otelgrpc.WithTracerProvider(tp))
usi := otelgrpc.UnaryServerInterceptor(otelgrpc.WithTracerProvider(tp), otelgrpc.WithPeerAddrAsAttribute(true))
runServerChecks(t, usi, sr)
}

// TestUnaryServerInterceptor tests the server interceptor for unary RPCs.
func TestUnaryServerInterceptorWithoutPeerAddrAsAttribute(t *testing.T) {
sr := tracetest.NewSpanRecorder()
tp := trace.NewTracerProvider(trace.WithSpanProcessor(sr))
usi := otelgrpc.UnaryServerInterceptor(otelgrpc.WithTracerProvider(tp), otelgrpc.WithPeerAddrAsAttribute(false))
runServerChecks(t, usi, sr)
}

func runServerChecks(t *testing.T, usi grpc.UnaryServerInterceptor, sr *tracetest.SpanRecorder) {
for _, check := range serverChecks {
name := check.grpcCode.String()
t.Run(name, func(t *testing.T) {
Expand Down

0 comments on commit caa8291

Please sign in to comment.