chore(deps): bump github.com/containerd/containerd from 1.6.18 to 1.7.13 (#6102)

Signed-off-by: hongzhouzi <hongzhouzi@kubesphere.io>

vendor/go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc/config.go

@@ -15,14 +15,11 @@
 package otelgrpc // import "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 
 import (
-	"context"
-
-	"google.golang.org/grpc/metadata"
-
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/baggage"
+	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/propagation"
+	semconv "go.opentelemetry.io/otel/semconv/v1.17.0"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -40,9 +37,17 @@ type Filter func(*InterceptorInfo) bool
 
 // config is a group of options for this instrumentation.
 type config struct {
-	Filter         Filter
-	Propagators    propagation.TextMapPropagator
-	TracerProvider trace.TracerProvider
+	Filter           Filter
+	Propagators      propagation.TextMapPropagator
+	TracerProvider   trace.TracerProvider
+	MeterProvider    metric.MeterProvider
+	SpanStartOptions []trace.SpanStartOption
+
+	ReceivedEvent bool
+	SentEvent     bool
+
+	meter             metric.Meter
+	rpcServerDuration metric.Int64Histogram
 }
 
 // Option applies an option value for a config.
@@ -55,10 +60,25 @@ func newConfig(opts []Option) *config {
 	c := &config{
 		Propagators:    otel.GetTextMapPropagator(),
 		TracerProvider: otel.GetTracerProvider(),
+		MeterProvider:  otel.GetMeterProvider(),
 	}
 	for _, o := range opts {
 		o.apply(c)
 	}
+
+	c.meter = c.MeterProvider.Meter(
+		instrumentationName,
+		metric.WithInstrumentationVersion(Version()),
+		metric.WithSchemaURL(semconv.SchemaURL),
+	)
+	var err error
+	c.rpcServerDuration, err = c.meter.Int64Histogram("rpc.server.duration",
+		metric.WithDescription("Measures the duration of inbound RPC."),
+		metric.WithUnit("ms"))
+	if err != nil {
+		otel.Handle(err)
+	}
+
 	return c
 }
 
@@ -105,59 +125,63 @@ func WithTracerProvider(tp trace.TracerProvider) Option {
 	return tracerProviderOption{tp: tp}
 }
 
-type metadataSupplier struct {
-	metadata *metadata.MD
-}
+type meterProviderOption struct{ mp metric.MeterProvider }
 
-// assert that metadataSupplier implements the TextMapCarrier interface.
-var _ propagation.TextMapCarrier = &metadataSupplier{}
-
-func (s *metadataSupplier) Get(key string) string {
-	values := s.metadata.Get(key)
-	if len(values) == 0 {
-		return ""
+func (o meterProviderOption) apply(c *config) {
+	if o.mp != nil {
+		c.MeterProvider = o.mp
 	}
-	return values[0]
 }
 
-func (s *metadataSupplier) Set(key string, value string) {
-	s.metadata.Set(key, value)
+// 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 {
+	return meterProviderOption{mp: mp}
 }
 
-func (s *metadataSupplier) Keys() []string {
-	out := make([]string, 0, len(*s.metadata))
-	for key := range *s.metadata {
-		out = append(out, key)
+// Event type that can be recorded, see WithMessageEvents.
+type Event int
+
+// Different types of events that can be recorded, see WithMessageEvents.
+const (
+	ReceivedEvents Event = iota
+	SentEvents
+)
+
+type messageEventsProviderOption struct {
+	events []Event
+}
+
+func (m messageEventsProviderOption) apply(c *config) {
+	for _, e := range m.events {
+		switch e {
+		case ReceivedEvents:
+			c.ReceivedEvent = true
+		case SentEvents:
+			c.SentEvent = true
+		}
 	}
-	return out
 }
 
-// Inject injects correlation context and span context into the gRPC
-// metadata object. This function is meant to be used on outgoing
-// requests.
-func Inject(ctx context.Context, md *metadata.MD, opts ...Option) {
-	c := newConfig(opts)
-	inject(ctx, md, c.Propagators)
+// WithMessageEvents configures the Handler to record the specified events
+// (span.AddEvent) on spans. By default only summary attributes are added at the
+// end of the request.
+//
+// Valid events are:
+//   - ReceivedEvents: Record the number of bytes read after every gRPC read operation.
+//   - SentEvents: Record the number of bytes written after every gRPC write operation.
+func WithMessageEvents(events ...Event) Option {
+	return messageEventsProviderOption{events: events}
 }
 
-func inject(ctx context.Context, md *metadata.MD, propagators propagation.TextMapPropagator) {
-	propagators.Inject(ctx, &metadataSupplier{
-		metadata: md,
-	})
+type spanStartOption struct{ opts []trace.SpanStartOption }
+
+func (o spanStartOption) apply(c *config) {
+	c.SpanStartOptions = append(c.SpanStartOptions, o.opts...)
 }
 
-// Extract returns the correlation context and span context that
-// another service encoded in the gRPC metadata object with Inject.
-// This function is meant to be used on incoming requests.
-func Extract(ctx context.Context, md *metadata.MD, opts ...Option) (baggage.Baggage, trace.SpanContext) {
-	c := newConfig(opts)
-	return extract(ctx, md, c.Propagators)
-}
-
-func extract(ctx context.Context, md *metadata.MD, propagators propagation.TextMapPropagator) (baggage.Baggage, trace.SpanContext) {
-	ctx = propagators.Extract(ctx, &metadataSupplier{
-		metadata: md,
-	})
-
-	return baggage.FromContext(ctx), trace.SpanContextFromContext(ctx)
+// WithSpanOptions configures an additional set of
+// trace.SpanOptions, which are applied to each new span.
+func WithSpanOptions(opts ...trace.SpanStartOption) Option {
+	return spanStartOption{opts}
 }

vendor/go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc/doc.go

@@ -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"

vendor/go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc/interceptor.go

@@ -20,41 +20,37 @@ import (
 	"context"
 	"io"
 	"net"
-
-	"github.com/golang/protobuf/proto" // nolint:staticcheck
+	"strconv"
+	"time"
 
 	"google.golang.org/grpc"
 	grpc_codes "google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc/internal"
 	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/baggage"
 	"go.opentelemetry.io/otel/codes"
-	semconv "go.opentelemetry.io/otel/semconv/v1.12.0"
+	"go.opentelemetry.io/otel/metric"
+	semconv "go.opentelemetry.io/otel/semconv/v1.17.0"
 	"go.opentelemetry.io/otel/trace"
 )
 
 type messageType attribute.KeyValue
 
 // Event adds an event of the messageType to the span associated with the
-// passed context with id and size (if message is a proto message).
-func (m messageType) Event(ctx context.Context, id int, message interface{}) {
+// passed context with a message id.
+func (m messageType) Event(ctx context.Context, id int, _ interface{}) {
 	span := trace.SpanFromContext(ctx)
-	if p, ok := message.(proto.Message); ok {
-		span.AddEvent("message", trace.WithAttributes(
-			attribute.KeyValue(m),
-			RPCMessageIDKey.Int(id),
-			RPCMessageUncompressedSizeKey.Int(proto.Size(p)),
-		))
-	} else {
-		span.AddEvent("message", trace.WithAttributes(
-			attribute.KeyValue(m),
-			RPCMessageIDKey.Int(id),
-		))
+	if !span.IsRecording() {
+		return
 	}
+	span.AddEvent("message", trace.WithAttributes(
+		attribute.KeyValue(m),
+		RPCMessageIDKey.Int(id),
+	))
 }
 
 var (
@@ -66,6 +62,11 @@ var (
 // for use in a grpc.Dial call.
 func UnaryClientInterceptor(opts ...Option) grpc.UnaryClientInterceptor {
 	cfg := newConfig(opts)
+	tracer := cfg.TracerProvider.Tracer(
+		instrumentationName,
+		trace.WithInstrumentationVersion(Version()),
+	)
+
 	return func(
 		ctx context.Context,
 		method string,
@@ -82,32 +83,32 @@ func UnaryClientInterceptor(opts ...Option) grpc.UnaryClientInterceptor {
 			return invoker(ctx, method, req, reply, cc, callOpts...)
 		}
 
-		requestMetadata, _ := metadata.FromOutgoingContext(ctx)
-		metadataCopy := requestMetadata.Copy()
+		name, attr := spanInfo(method, cc.Target())
 
-		tracer := cfg.TracerProvider.Tracer(
-			instrumentationName,
-			trace.WithInstrumentationVersion(SemVersion()),
+		startOpts := append([]trace.SpanStartOption{
+			trace.WithSpanKind(trace.SpanKindClient),
+			trace.WithAttributes(attr...)},
+			cfg.SpanStartOptions...,
 		)
 
-		name, attr := spanInfo(method, cc.Target())
-		var span trace.Span
-		ctx, span = tracer.Start(
+		ctx, span := tracer.Start(
 			ctx,
 			name,
-			trace.WithSpanKind(trace.SpanKindClient),
-			trace.WithAttributes(attr...),
+			startOpts...,
 		)
 		defer span.End()
 
-		inject(ctx, &metadataCopy, cfg.Propagators)
-		ctx = metadata.NewOutgoingContext(ctx, metadataCopy)
+		ctx = inject(ctx, cfg.Propagators)
 
-		messageSent.Event(ctx, 1, req)
+		if cfg.SentEvent {
+			messageSent.Event(ctx, 1, req)
+		}
 
 		err := invoker(ctx, method, req, reply, cc, callOpts...)
 
-		messageReceived.Event(ctx, 1, reply)
+		if cfg.ReceivedEvent {
+			messageReceived.Event(ctx, 1, reply)
+		}
 
 		if err != nil {
 			s, _ := status.FromError(err)
@@ -143,6 +144,9 @@ type clientStream struct {
 	eventsDone chan struct{}
 	finished   chan error
 
+	receivedEvent bool
+	sentEvent     bool
+
 	receivedMessageID int
 	sentMessageID     int
 }
@@ -160,7 +164,10 @@ func (w *clientStream) RecvMsg(m interface{}) error {
 		w.sendStreamEvent(errorEvent, err)
 	} else {
 		w.receivedMessageID++
-		messageReceived.Event(w.Context(), w.receivedMessageID, m)
+
+		if w.receivedEvent {
+			messageReceived.Event(w.Context(), w.receivedMessageID, m)
+		}
 	}
 
 	return err
@@ -170,7 +177,10 @@ func (w *clientStream) SendMsg(m interface{}) error {
 	err := w.ClientStream.SendMsg(m)
 
 	w.sentMessageID++
-	messageSent.Event(w.Context(), w.sentMessageID, m)
+
+	if w.sentEvent {
+		messageSent.Event(w.Context(), w.sentMessageID, m)
+	}
 
 	if err != nil {
 		w.sendStreamEvent(errorEvent, err)
@@ -199,7 +209,7 @@ func (w *clientStream) CloseSend() error {
 	return err
 }
 
-func wrapClientStream(ctx context.Context, s grpc.ClientStream, desc *grpc.StreamDesc) *clientStream {
+func wrapClientStream(ctx context.Context, s grpc.ClientStream, desc *grpc.StreamDesc, cfg *config) *clientStream {
 	events := make(chan streamEvent)
 	eventsDone := make(chan struct{})
 	finished := make(chan error)
@@ -226,11 +236,13 @@ func wrapClientStream(ctx context.Context, s grpc.ClientStream, desc *grpc.Strea
 	}()
 
 	return &clientStream{
-		ClientStream: s,
-		desc:         desc,
-		events:       events,
-		eventsDone:   eventsDone,
-		finished:     finished,
+		ClientStream:  s,
+		desc:          desc,
+		events:        events,
+		eventsDone:    eventsDone,
+		finished:      finished,
+		receivedEvent: cfg.ReceivedEvent,
+		sentEvent:     cfg.SentEvent,
 	}
 }
 
@@ -245,6 +257,11 @@ func (w *clientStream) sendStreamEvent(eventType streamEventType, err error) {
 // for use in a grpc.Dial call.
 func StreamClientInterceptor(opts ...Option) grpc.StreamClientInterceptor {
 	cfg := newConfig(opts)
+	tracer := cfg.TracerProvider.Tracer(
+		instrumentationName,
+		trace.WithInstrumentationVersion(Version()),
+	)
+
 	return func(
 		ctx context.Context,
 		desc *grpc.StreamDesc,
@@ -261,25 +278,21 @@ func StreamClientInterceptor(opts ...Option) grpc.StreamClientInterceptor {
 			return streamer(ctx, desc, cc, method, callOpts...)
 		}
 
-		requestMetadata, _ := metadata.FromOutgoingContext(ctx)
-		metadataCopy := requestMetadata.Copy()
+		name, attr := spanInfo(method, cc.Target())
 
-		tracer := cfg.TracerProvider.Tracer(
-			instrumentationName,
-			trace.WithInstrumentationVersion(SemVersion()),
+		startOpts := append([]trace.SpanStartOption{
+			trace.WithSpanKind(trace.SpanKindClient),
+			trace.WithAttributes(attr...)},
+			cfg.SpanStartOptions...,
 		)
 
-		name, attr := spanInfo(method, cc.Target())
-		var span trace.Span
-		ctx, span = tracer.Start(
+		ctx, span := tracer.Start(
 			ctx,
 			name,
-			trace.WithSpanKind(trace.SpanKindClient),
-			trace.WithAttributes(attr...),
+			startOpts...,
 		)
 
-		inject(ctx, &metadataCopy, cfg.Propagators)
-		ctx = metadata.NewOutgoingContext(ctx, metadataCopy)
+		ctx = inject(ctx, cfg.Propagators)
 
 		s, err := streamer(ctx, desc, cc, method, callOpts...)
 		if err != nil {
@@ -289,7 +302,7 @@ func StreamClientInterceptor(opts ...Option) grpc.StreamClientInterceptor {
 			span.End()
 			return s, err
 		}
-		stream := wrapClientStream(ctx, s, desc)
+		stream := wrapClientStream(ctx, s, desc, cfg)
 
 		go func() {
 			err := <-stream.finished
@@ -313,6 +326,11 @@ func StreamClientInterceptor(opts ...Option) grpc.StreamClientInterceptor {
 // for use in a grpc.NewServer call.
 func UnaryServerInterceptor(opts ...Option) grpc.UnaryServerInterceptor {
 	cfg := newConfig(opts)
+	tracer := cfg.TracerProvider.Tracer(
+		instrumentationName,
+		trace.WithInstrumentationVersion(Version()),
+	)
+
 	return func(
 		ctx context.Context,
 		req interface{},
@@ -327,37 +345,49 @@ func UnaryServerInterceptor(opts ...Option) grpc.UnaryServerInterceptor {
 			return handler(ctx, req)
 		}
 
-		requestMetadata, _ := metadata.FromIncomingContext(ctx)
-		metadataCopy := requestMetadata.Copy()
+		ctx = extract(ctx, cfg.Propagators)
+		name, attr := spanInfo(info.FullMethod, peerFromCtx(ctx))
 
-		bags, spanCtx := Extract(ctx, &metadataCopy, opts...)
-		ctx = baggage.ContextWithBaggage(ctx, bags)
-
-		tracer := cfg.TracerProvider.Tracer(
-			instrumentationName,
-			trace.WithInstrumentationVersion(SemVersion()),
+		startOpts := append([]trace.SpanStartOption{
+			trace.WithSpanKind(trace.SpanKindServer),
+			trace.WithAttributes(attr...)},
+			cfg.SpanStartOptions...,
 		)
 
-		name, attr := spanInfo(info.FullMethod, peerFromCtx(ctx))
 		ctx, span := tracer.Start(
-			trace.ContextWithRemoteSpanContext(ctx, spanCtx),
+			trace.ContextWithRemoteSpanContext(ctx, trace.SpanContextFromContext(ctx)),
 			name,
-			trace.WithSpanKind(trace.SpanKindServer),
-			trace.WithAttributes(attr...),
+			startOpts...,
 		)
 		defer span.End()
 
-		messageReceived.Event(ctx, 1, req)
+		if cfg.ReceivedEvent {
+			messageReceived.Event(ctx, 1, req)
+		}
+
+		var statusCode grpc_codes.Code
+		defer func(t time.Time) {
+			elapsedTime := time.Since(t) / time.Millisecond
+			attr = append(attr, semconv.RPCGRPCStatusCodeKey.Int64(int64(statusCode)))
+			o := metric.WithAttributes(attr...)
+			cfg.rpcServerDuration.Record(ctx, int64(elapsedTime), o)
+		}(time.Now())
 
 		resp, err := handler(ctx, req)
 		if err != nil {
 			s, _ := status.FromError(err)
-			span.SetStatus(codes.Error, s.Message())
+			statusCode, msg := serverStatus(s)
+			span.SetStatus(statusCode, msg)
 			span.SetAttributes(statusCodeAttr(s.Code()))
-			messageSent.Event(ctx, 1, s.Proto())
+			if cfg.SentEvent {
+				messageSent.Event(ctx, 1, s.Proto())
+			}
 		} else {
+			statusCode = grpc_codes.OK
 			span.SetAttributes(statusCodeAttr(grpc_codes.OK))
-			messageSent.Event(ctx, 1, resp)
+			if cfg.SentEvent {
+				messageSent.Event(ctx, 1, resp)
+			}
 		}
 
 		return resp, err
@@ -372,6 +402,9 @@ type serverStream struct {
 
 	receivedMessageID int
 	sentMessageID     int
+
+	receivedEvent bool
+	sentEvent     bool
 }
 
 func (w *serverStream) Context() context.Context {
@@ -383,7 +416,9 @@ func (w *serverStream) RecvMsg(m interface{}) error {
 
 	if err == nil {
 		w.receivedMessageID++
-		messageReceived.Event(w.Context(), w.receivedMessageID, m)
+		if w.receivedEvent {
+			messageReceived.Event(w.Context(), w.receivedMessageID, m)
+		}
 	}
 
 	return err
@@ -393,15 +428,19 @@ func (w *serverStream) SendMsg(m interface{}) error {
 	err := w.ServerStream.SendMsg(m)
 
 	w.sentMessageID++
-	messageSent.Event(w.Context(), w.sentMessageID, m)
+	if w.sentEvent {
+		messageSent.Event(w.Context(), w.sentMessageID, m)
+	}
 
 	return err
 }
 
-func wrapServerStream(ctx context.Context, ss grpc.ServerStream) *serverStream {
+func wrapServerStream(ctx context.Context, ss grpc.ServerStream, cfg *config) *serverStream {
 	return &serverStream{
-		ServerStream: ss,
-		ctx:          ctx,
+		ServerStream:  ss,
+		ctx:           ctx,
+		receivedEvent: cfg.ReceivedEvent,
+		sentEvent:     cfg.SentEvent,
 	}
 }
 
@@ -409,6 +448,11 @@ func wrapServerStream(ctx context.Context, ss grpc.ServerStream) *serverStream {
 // for use in a grpc.NewServer call.
 func StreamServerInterceptor(opts ...Option) grpc.StreamServerInterceptor {
 	cfg := newConfig(opts)
+	tracer := cfg.TracerProvider.Tracer(
+		instrumentationName,
+		trace.WithInstrumentationVersion(Version()),
+	)
+
 	return func(
 		srv interface{},
 		ss grpc.ServerStream,
@@ -421,34 +465,30 @@ func StreamServerInterceptor(opts ...Option) grpc.StreamServerInterceptor {
 			Type:             StreamServer,
 		}
 		if cfg.Filter != nil && !cfg.Filter(i) {
-			return handler(srv, wrapServerStream(ctx, ss))
+			return handler(srv, wrapServerStream(ctx, ss, cfg))
 		}
 
-		requestMetadata, _ := metadata.FromIncomingContext(ctx)
-		metadataCopy := requestMetadata.Copy()
+		ctx = extract(ctx, cfg.Propagators)
+		name, attr := spanInfo(info.FullMethod, peerFromCtx(ctx))
 
-		bags, spanCtx := Extract(ctx, &metadataCopy, opts...)
-		ctx = baggage.ContextWithBaggage(ctx, bags)
-
-		tracer := cfg.TracerProvider.Tracer(
-			instrumentationName,
-			trace.WithInstrumentationVersion(SemVersion()),
+		startOpts := append([]trace.SpanStartOption{
+			trace.WithSpanKind(trace.SpanKindServer),
+			trace.WithAttributes(attr...)},
+			cfg.SpanStartOptions...,
 		)
 
-		name, attr := spanInfo(info.FullMethod, peerFromCtx(ctx))
 		ctx, span := tracer.Start(
-			trace.ContextWithRemoteSpanContext(ctx, spanCtx),
+			trace.ContextWithRemoteSpanContext(ctx, trace.SpanContextFromContext(ctx)),
 			name,
-			trace.WithSpanKind(trace.SpanKindServer),
-			trace.WithAttributes(attr...),
+			startOpts...,
 		)
 		defer span.End()
 
-		err := handler(srv, wrapServerStream(ctx, ss))
-
+		err := handler(srv, wrapServerStream(ctx, ss, cfg))
 		if err != nil {
 			s, _ := status.FromError(err)
-			span.SetStatus(codes.Error, s.Message())
+			statusCode, msg := serverStatus(s)
+			span.SetStatus(statusCode, msg)
 			span.SetAttributes(statusCodeAttr(s.Code()))
 		} else {
 			span.SetAttributes(statusCodeAttr(grpc_codes.OK))
@@ -461,28 +501,45 @@ func StreamServerInterceptor(opts ...Option) grpc.StreamServerInterceptor {
 // spanInfo returns a span name and all appropriate attributes from the gRPC
 // method and peer address.
 func spanInfo(fullMethod, peerAddress string) (string, []attribute.KeyValue) {
-	attrs := []attribute.KeyValue{RPCSystemGRPC}
 	name, mAttrs := internal.ParseFullMethod(fullMethod)
+	peerAttrs := peerAttr(peerAddress)
+
+	attrs := make([]attribute.KeyValue, 0, 1+len(mAttrs)+len(peerAttrs))
+	attrs = append(attrs, RPCSystemGRPC)
 	attrs = append(attrs, mAttrs...)
-	attrs = append(attrs, peerAttr(peerAddress)...)
+	attrs = append(attrs, peerAttrs...)
 	return name, attrs
 }
 
 // peerAttr returns attributes about the peer address.
 func peerAttr(addr string) []attribute.KeyValue {
-	host, port, err := net.SplitHostPort(addr)
+	host, p, err := net.SplitHostPort(addr)
 	if err != nil {
-		return []attribute.KeyValue(nil)
+		return nil
 	}
 
 	if host == "" {
 		host = "127.0.0.1"
 	}
-
-	return []attribute.KeyValue{
-		semconv.NetPeerIPKey.String(host),
-		semconv.NetPeerPortKey.String(port),
+	port, err := strconv.Atoi(p)
+	if err != nil {
+		return nil
 	}
+
+	var attr []attribute.KeyValue
+	if ip := net.ParseIP(host); ip != nil {
+		attr = []attribute.KeyValue{
+			semconv.NetSockPeerAddr(host),
+			semconv.NetSockPeerPort(port),
+		}
+	} else {
+		attr = []attribute.KeyValue{
+			semconv.NetPeerName(host),
+			semconv.NetPeerPort(port),
+		}
+	}
+
+	return attr
 }
 
 // peerFromCtx returns a peer address from a context, if one exists.
@@ -498,3 +555,26 @@ func peerFromCtx(ctx context.Context) string {
 func statusCodeAttr(c grpc_codes.Code) attribute.KeyValue {
 	return GRPCStatusCodeKey.Int64(int64(c))
 }
+
+// serverStatus returns a span status code and message for a given gRPC
+// status code. It maps specific gRPC status codes to a corresponding span
+// status code and message. This function is intended for use on the server
+// side of a gRPC connection.
+//
+// If the gRPC status code is Unknown, DeadlineExceeded, Unimplemented,
+// Internal, Unavailable, or DataLoss, it returns a span status code of Error
+// and the message from the gRPC status. Otherwise, it returns a span status
+// code of Unset and an empty message.
+func serverStatus(grpcStatus *status.Status) (codes.Code, string) {
+	switch grpcStatus.Code() {
+	case grpc_codes.Unknown,
+		grpc_codes.DeadlineExceeded,
+		grpc_codes.Unimplemented,
+		grpc_codes.Internal,
+		grpc_codes.Unavailable,
+		grpc_codes.DataLoss:
+		return codes.Error, grpcStatus.Message()
+	default:
+		return codes.Unset, ""
+	}
+}

vendor/go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc/internal/parse.go

@@ -18,26 +18,34 @@ import (
 	"strings"
 
 	"go.opentelemetry.io/otel/attribute"
-	semconv "go.opentelemetry.io/otel/semconv/v1.12.0"
+	semconv "go.opentelemetry.io/otel/semconv/v1.17.0"
 )
 
 // ParseFullMethod returns a span name following the OpenTelemetry semantic
 // conventions as well as all applicable span attribute.KeyValue attributes based
 // on a gRPC's FullMethod.
+//
+// Parsing is consistent with grpc-go implementation:
+// https://github.com/grpc/grpc-go/blob/v1.57.0/internal/grpcutil/method.go#L26-L39
 func ParseFullMethod(fullMethod string) (string, []attribute.KeyValue) {
-	name := strings.TrimLeft(fullMethod, "/")
-	parts := strings.SplitN(name, "/", 2)
-	if len(parts) != 2 {
+	if !strings.HasPrefix(fullMethod, "/") {
 		// Invalid format, does not follow `/package.service/method`.
-		return name, []attribute.KeyValue(nil)
+		return fullMethod, nil
 	}
+	name := fullMethod[1:]
+	pos := strings.LastIndex(name, "/")
+	if pos < 0 {
+		// Invalid format, does not follow `/package.service/method`.
+		return name, nil
+	}
+	service, method := name[:pos], name[pos+1:]
 
 	var attrs []attribute.KeyValue
-	if service := parts[0]; service != "" {
-		attrs = append(attrs, semconv.RPCServiceKey.String(service))
+	if service != "" {
+		attrs = append(attrs, semconv.RPCService(service))
 	}
-	if method := parts[1]; method != "" {
-		attrs = append(attrs, semconv.RPCMethodKey.String(method))
+	if method != "" {
+		attrs = append(attrs, semconv.RPCMethod(method))
 	}
 	return name, attrs
 }

vendor/go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc/metadata_supplier.go

@@ -0,0 +1,98 @@
+// 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"
+
+	"google.golang.org/grpc/metadata"
+
+	"go.opentelemetry.io/otel/baggage"
+	"go.opentelemetry.io/otel/propagation"
+	"go.opentelemetry.io/otel/trace"
+)
+
+type metadataSupplier struct {
+	metadata *metadata.MD
+}
+
+// assert that metadataSupplier implements the TextMapCarrier interface.
+var _ propagation.TextMapCarrier = &metadataSupplier{}
+
+func (s *metadataSupplier) Get(key string) string {
+	values := s.metadata.Get(key)
+	if len(values) == 0 {
+		return ""
+	}
+	return values[0]
+}
+
+func (s *metadataSupplier) Set(key string, value string) {
+	s.metadata.Set(key, value)
+}
+
+func (s *metadataSupplier) Keys() []string {
+	out := make([]string, 0, len(*s.metadata))
+	for key := range *s.metadata {
+		out = append(out, key)
+	}
+	return out
+}
+
+// Inject injects correlation context and span context into the gRPC
+// metadata object. This function is meant to be used on outgoing
+// requests.
+// Deprecated: Unnecessary public func.
+func Inject(ctx context.Context, md *metadata.MD, opts ...Option) {
+	c := newConfig(opts)
+	c.Propagators.Inject(ctx, &metadataSupplier{
+		metadata: md,
+	})
+}
+
+func inject(ctx context.Context, propagators propagation.TextMapPropagator) context.Context {
+	md, ok := metadata.FromOutgoingContext(ctx)
+	if !ok {
+		md = metadata.MD{}
+	}
+	propagators.Inject(ctx, &metadataSupplier{
+		metadata: &md,
+	})
+	return metadata.NewOutgoingContext(ctx, md)
+}
+
+// Extract returns the correlation context and span context that
+// another service encoded in the gRPC metadata object with Inject.
+// This function is meant to be used on incoming requests.
+// Deprecated: Unnecessary public func.
+func Extract(ctx context.Context, md *metadata.MD, opts ...Option) (baggage.Baggage, trace.SpanContext) {
+	c := newConfig(opts)
+	ctx = c.Propagators.Extract(ctx, &metadataSupplier{
+		metadata: md,
+	})
+
+	return baggage.FromContext(ctx), trace.SpanContextFromContext(ctx)
+}
+
+func extract(ctx context.Context, propagators propagation.TextMapPropagator) context.Context {
+	md, ok := metadata.FromIncomingContext(ctx)
+	if !ok {
+		md = metadata.MD{}
+	}
+
+	return propagators.Extract(ctx, &metadataSupplier{
+		metadata: &md,
+	})
+}

vendor/go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc/semconv.go

@@ -16,7 +16,7 @@ package otelgrpc // import "go.opentelemetry.io/contrib/instrumentation/google.g
 
 import (
 	"go.opentelemetry.io/otel/attribute"
-	semconv "go.opentelemetry.io/otel/semconv/v1.12.0"
+	semconv "go.opentelemetry.io/otel/semconv/v1.17.0"
 )
 
 // Semantic conventions for attribute keys for gRPC.
@@ -41,7 +41,7 @@ const (
 // Semantic conventions for common RPC attributes.
 var (
 	// Semantic convention for gRPC as the remoting system.
-	RPCSystemGRPC = semconv.RPCSystemKey.String("grpc")
+	RPCSystemGRPC = semconv.RPCSystemGRPC
 
 	// Semantic convention for a message named message.
 	RPCNameMessage = RPCNameKey.String("message")

vendor/go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc/stats_handler.go

@@ -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
+	}
+}

vendor/go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc/version.go

@@ -16,11 +16,13 @@ package otelgrpc // import "go.opentelemetry.io/contrib/instrumentation/google.g
 
 // Version is the current release version of the gRPC instrumentation.
 func Version() string {
-	return "0.35.0"
+	return "0.45.0"
 	// This string is updated by the pre_release.sh script during release
 }
 
 // SemVersion is the semantic version to be supplied to tracer/meter creation.
+//
+// Deprecated: Use [Version] instead.
 func SemVersion() string {
-	return "semver:" + Version()
+	return Version()
 }