vendor: github.com/moby/buildkit v0.13.0-rc2

full diff: https://github.com/moby/buildkit/compare/8e3fe35738c2...v0.13.0-rc2

Signed-off-by: CrazyMax <1951866+crazy-max@users.noreply.github.com>

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

@@ -24,8 +24,8 @@ import (
 )
 
 const (
-	// instrumentationName is the name of this instrumentation package.
-	instrumentationName = "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	// ScopeName is the instrumentation scope name.
+	ScopeName = "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	// GRPCStatusCodeKey is convention for numeric status code of a gRPC request.
 	GRPCStatusCodeKey = attribute.Key("rpc.grpc.status_code")
 )
@@ -46,8 +46,14 @@ type config struct {
 	ReceivedEvent bool
 	SentEvent     bool
 
-	meter             metric.Meter
-	rpcServerDuration metric.Int64Histogram
+	tracer trace.Tracer
+	meter  metric.Meter
+
+	rpcDuration        metric.Float64Histogram
+	rpcRequestSize     metric.Int64Histogram
+	rpcResponseSize    metric.Int64Histogram
+	rpcRequestsPerRPC  metric.Int64Histogram
+	rpcResponsesPerRPC metric.Int64Histogram
 }
 
 // Option applies an option value for a config.
@@ -56,7 +62,7 @@ type Option interface {
 }
 
 // newConfig returns a config configured with all the passed Options.
-func newConfig(opts []Option) *config {
+func newConfig(opts []Option, role string) *config {
 	c := &config{
 		Propagators:    otel.GetTextMapPropagator(),
 		TracerProvider: otel.GetTracerProvider(),
@@ -66,19 +72,53 @@ func newConfig(opts []Option) *config {
 		o.apply(c)
 	}
 
+	c.tracer = c.TracerProvider.Tracer(
+		ScopeName,
+		trace.WithInstrumentationVersion(SemVersion()),
+	)
+
 	c.meter = c.MeterProvider.Meter(
-		instrumentationName,
+		ScopeName,
 		metric.WithInstrumentationVersion(Version()),
 		metric.WithSchemaURL(semconv.SchemaURL),
 	)
+
 	var err error
-	c.rpcServerDuration, err = c.meter.Int64Histogram("rpc.server.duration",
+	c.rpcDuration, err = c.meter.Float64Histogram("rpc."+role+".duration",
 		metric.WithDescription("Measures the duration of inbound RPC."),
 		metric.WithUnit("ms"))
 	if err != nil {
 		otel.Handle(err)
 	}
 
+	c.rpcRequestSize, err = c.meter.Int64Histogram("rpc."+role+".request.size",
+		metric.WithDescription("Measures size of RPC request messages (uncompressed)."),
+		metric.WithUnit("By"))
+	if err != nil {
+		otel.Handle(err)
+	}
+
+	c.rpcResponseSize, err = c.meter.Int64Histogram("rpc."+role+".response.size",
+		metric.WithDescription("Measures size of RPC response messages (uncompressed)."),
+		metric.WithUnit("By"))
+	if err != nil {
+		otel.Handle(err)
+	}
+
+	c.rpcRequestsPerRPC, err = c.meter.Int64Histogram("rpc."+role+".requests_per_rpc",
+		metric.WithDescription("Measures the number of messages received per RPC. Should be 1 for all non-streaming RPCs."),
+		metric.WithUnit("{count}"))
+	if err != nil {
+		otel.Handle(err)
+	}
+
+	c.rpcResponsesPerRPC, err = c.meter.Int64Histogram("rpc."+role+".responses_per_rpc",
+		metric.WithDescription("Measures the number of messages received per RPC. Should be 1 for all non-streaming RPCs."),
+		metric.WithUnit("{count}"))
+	if err != nil {
+		otel.Handle(err)
+	}
+
 	return c
 }
 
@@ -105,6 +145,8 @@ func (o tracerProviderOption) apply(c *config) {
 }
 
 // WithInterceptorFilter returns an Option to use the request filter.
+//
+// Deprecated: Use stats handlers instead.
 func WithInterceptorFilter(f Filter) Option {
 	return interceptorFilterOption{f: f}
 }

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

@@ -13,33 +13,10 @@
 // limitations under the License.
 
 /*
-Package otelgrpc is the instrumentation library for [google.golang.org/grpc]
+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:
+Use [NewClientHandler] with [grpc.WithStatsHandler] to instrument a gRPC client.
 
-  - 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
+Use [NewServerHandler] with [grpc.StatsHandler] to instrument a gRPC server.
 */
 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

@@ -60,10 +60,12 @@ var (
 
 // UnaryClientInterceptor returns a grpc.UnaryClientInterceptor suitable
 // for use in a grpc.Dial call.
+//
+// Deprecated: Use [NewClientHandler] instead.
 func UnaryClientInterceptor(opts ...Option) grpc.UnaryClientInterceptor {
-	cfg := newConfig(opts)
+	cfg := newConfig(opts, "client")
 	tracer := cfg.TracerProvider.Tracer(
-		instrumentationName,
+		ScopeName,
 		trace.WithInstrumentationVersion(Version()),
 	)
 
@@ -83,11 +85,12 @@ func UnaryClientInterceptor(opts ...Option) grpc.UnaryClientInterceptor {
 			return invoker(ctx, method, req, reply, cc, callOpts...)
 		}
 
-		name, attr := spanInfo(method, cc.Target())
+		name, attr, _ := telemetryAttributes(method, cc.Target())
 
 		startOpts := append([]trace.SpanStartOption{
 			trace.WithSpanKind(trace.SpanKindClient),
-			trace.WithAttributes(attr...)},
+			trace.WithAttributes(attr...),
+		},
 			cfg.SpanStartOptions...,
 		)
 
@@ -122,27 +125,13 @@ func UnaryClientInterceptor(opts ...Option) grpc.UnaryClientInterceptor {
 	}
 }
 
-type streamEventType int
-
-type streamEvent struct {
-	Type streamEventType
-	Err  error
-}
-
-const (
-	receiveEndEvent streamEventType = iota
-	errorEvent
-)
-
 // clientStream  wraps around the embedded grpc.ClientStream, and intercepts the RecvMsg and
 // SendMsg method call.
 type clientStream struct {
 	grpc.ClientStream
+	desc *grpc.StreamDesc
 
-	desc       *grpc.StreamDesc
-	events     chan streamEvent
-	eventsDone chan struct{}
-	finished   chan error
+	span trace.Span
 
 	receivedEvent bool
 	sentEvent     bool
@@ -157,11 +146,11 @@ func (w *clientStream) RecvMsg(m interface{}) error {
 	err := w.ClientStream.RecvMsg(m)
 
 	if err == nil && !w.desc.ServerStreams {
-		w.sendStreamEvent(receiveEndEvent, nil)
+		w.endSpan(nil)
 	} else if err == io.EOF {
-		w.sendStreamEvent(receiveEndEvent, nil)
+		w.endSpan(nil)
 	} else if err != nil {
-		w.sendStreamEvent(errorEvent, err)
+		w.endSpan(err)
 	} else {
 		w.receivedMessageID++
 
@@ -183,7 +172,7 @@ func (w *clientStream) SendMsg(m interface{}) error {
 	}
 
 	if err != nil {
-		w.sendStreamEvent(errorEvent, err)
+		w.endSpan(err)
 	}
 
 	return err
@@ -191,9 +180,8 @@ func (w *clientStream) SendMsg(m interface{}) error {
 
 func (w *clientStream) Header() (metadata.MD, error) {
 	md, err := w.ClientStream.Header()
-
 	if err != nil {
-		w.sendStreamEvent(errorEvent, err)
+		w.endSpan(err)
 	}
 
 	return md, err
@@ -201,64 +189,43 @@ func (w *clientStream) Header() (metadata.MD, error) {
 
 func (w *clientStream) CloseSend() error {
 	err := w.ClientStream.CloseSend()
-
 	if err != nil {
-		w.sendStreamEvent(errorEvent, err)
+		w.endSpan(err)
 	}
 
 	return err
 }
 
-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)
-
-	go func() {
-		defer close(eventsDone)
-
-		for {
-			select {
-			case event := <-events:
-				switch event.Type {
-				case receiveEndEvent:
-					finished <- nil
-					return
-				case errorEvent:
-					finished <- event.Err
-					return
-				}
-			case <-ctx.Done():
-				finished <- ctx.Err()
-				return
-			}
-		}
-	}()
-
+func wrapClientStream(ctx context.Context, s grpc.ClientStream, desc *grpc.StreamDesc, span trace.Span, cfg *config) *clientStream {
 	return &clientStream{
 		ClientStream:  s,
+		span:          span,
 		desc:          desc,
-		events:        events,
-		eventsDone:    eventsDone,
-		finished:      finished,
 		receivedEvent: cfg.ReceivedEvent,
 		sentEvent:     cfg.SentEvent,
 	}
 }
 
-func (w *clientStream) sendStreamEvent(eventType streamEventType, err error) {
-	select {
-	case <-w.eventsDone:
-	case w.events <- streamEvent{Type: eventType, Err: err}:
+func (w *clientStream) endSpan(err error) {
+	if err != nil {
+		s, _ := status.FromError(err)
+		w.span.SetStatus(codes.Error, s.Message())
+		w.span.SetAttributes(statusCodeAttr(s.Code()))
+	} else {
+		w.span.SetAttributes(statusCodeAttr(grpc_codes.OK))
 	}
+
+	w.span.End()
 }
 
 // StreamClientInterceptor returns a grpc.StreamClientInterceptor suitable
 // for use in a grpc.Dial call.
+//
+// Deprecated: Use [NewClientHandler] instead.
 func StreamClientInterceptor(opts ...Option) grpc.StreamClientInterceptor {
-	cfg := newConfig(opts)
+	cfg := newConfig(opts, "client")
 	tracer := cfg.TracerProvider.Tracer(
-		instrumentationName,
+		ScopeName,
 		trace.WithInstrumentationVersion(Version()),
 	)
 
@@ -278,11 +245,12 @@ func StreamClientInterceptor(opts ...Option) grpc.StreamClientInterceptor {
 			return streamer(ctx, desc, cc, method, callOpts...)
 		}
 
-		name, attr := spanInfo(method, cc.Target())
+		name, attr, _ := telemetryAttributes(method, cc.Target())
 
 		startOpts := append([]trace.SpanStartOption{
 			trace.WithSpanKind(trace.SpanKindClient),
-			trace.WithAttributes(attr...)},
+			trace.WithAttributes(attr...),
+		},
 			cfg.SpanStartOptions...,
 		)
 
@@ -302,32 +270,19 @@ func StreamClientInterceptor(opts ...Option) grpc.StreamClientInterceptor {
 			span.End()
 			return s, err
 		}
-		stream := wrapClientStream(ctx, s, desc, cfg)
-
-		go func() {
-			err := <-stream.finished
-
-			if err != nil {
-				s, _ := status.FromError(err)
-				span.SetStatus(codes.Error, s.Message())
-				span.SetAttributes(statusCodeAttr(s.Code()))
-			} else {
-				span.SetAttributes(statusCodeAttr(grpc_codes.OK))
-			}
-
-			span.End()
-		}()
-
+		stream := wrapClientStream(ctx, s, desc, span, cfg)
 		return stream, nil
 	}
 }
 
 // UnaryServerInterceptor returns a grpc.UnaryServerInterceptor suitable
 // for use in a grpc.NewServer call.
+//
+// Deprecated: Use [NewServerHandler] instead.
 func UnaryServerInterceptor(opts ...Option) grpc.UnaryServerInterceptor {
-	cfg := newConfig(opts)
+	cfg := newConfig(opts, "server")
 	tracer := cfg.TracerProvider.Tracer(
-		instrumentationName,
+		ScopeName,
 		trace.WithInstrumentationVersion(Version()),
 	)
 
@@ -346,11 +301,12 @@ func UnaryServerInterceptor(opts ...Option) grpc.UnaryServerInterceptor {
 		}
 
 		ctx = extract(ctx, cfg.Propagators)
-		name, attr := spanInfo(info.FullMethod, peerFromCtx(ctx))
+		name, attr, metricAttrs := telemetryAttributes(info.FullMethod, peerFromCtx(ctx))
 
 		startOpts := append([]trace.SpanStartOption{
 			trace.WithSpanKind(trace.SpanKindServer),
-			trace.WithAttributes(attr...)},
+			trace.WithAttributes(attr...),
+		},
 			cfg.SpanStartOptions...,
 		)
 
@@ -365,30 +321,30 @@ func UnaryServerInterceptor(opts ...Option) grpc.UnaryServerInterceptor {
 			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())
+		before := time.Now()
 
 		resp, err := handler(ctx, req)
+
+		s, _ := status.FromError(err)
 		if err != nil {
-			s, _ := status.FromError(err)
 			statusCode, msg := serverStatus(s)
 			span.SetStatus(statusCode, msg)
-			span.SetAttributes(statusCodeAttr(s.Code()))
 			if cfg.SentEvent {
 				messageSent.Event(ctx, 1, s.Proto())
 			}
 		} else {
-			statusCode = grpc_codes.OK
-			span.SetAttributes(statusCodeAttr(grpc_codes.OK))
 			if cfg.SentEvent {
 				messageSent.Event(ctx, 1, resp)
 			}
 		}
+		grpcStatusCodeAttr := statusCodeAttr(s.Code())
+		span.SetAttributes(grpcStatusCodeAttr)
+
+		// Use floating point division here for higher precision (instead of Millisecond method).
+		elapsedTime := float64(time.Since(before)) / float64(time.Millisecond)
+
+		metricAttrs = append(metricAttrs, grpcStatusCodeAttr)
+		cfg.rpcDuration.Record(ctx, elapsedTime, metric.WithAttributes(metricAttrs...))
 
 		return resp, err
 	}
@@ -446,10 +402,12 @@ func wrapServerStream(ctx context.Context, ss grpc.ServerStream, cfg *config) *s
 
 // StreamServerInterceptor returns a grpc.StreamServerInterceptor suitable
 // for use in a grpc.NewServer call.
+//
+// Deprecated: Use [NewServerHandler] instead.
 func StreamServerInterceptor(opts ...Option) grpc.StreamServerInterceptor {
-	cfg := newConfig(opts)
+	cfg := newConfig(opts, "server")
 	tracer := cfg.TracerProvider.Tracer(
-		instrumentationName,
+		ScopeName,
 		trace.WithInstrumentationVersion(Version()),
 	)
 
@@ -469,11 +427,12 @@ func StreamServerInterceptor(opts ...Option) grpc.StreamServerInterceptor {
 		}
 
 		ctx = extract(ctx, cfg.Propagators)
-		name, attr := spanInfo(info.FullMethod, peerFromCtx(ctx))
+		name, attr, _ := telemetryAttributes(info.FullMethod, peerFromCtx(ctx))
 
 		startOpts := append([]trace.SpanStartOption{
 			trace.WithSpanKind(trace.SpanKindServer),
-			trace.WithAttributes(attr...)},
+			trace.WithAttributes(attr...),
+		},
 			cfg.SpanStartOptions...,
 		)
 
@@ -498,17 +457,18 @@ 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) {
-	name, mAttrs := internal.ParseFullMethod(fullMethod)
+// telemetryAttributes returns a span name and span and metric attributes from
+// the gRPC method and peer address.
+func telemetryAttributes(fullMethod, peerAddress string) (string, []attribute.KeyValue, []attribute.KeyValue) {
+	name, methodAttrs := internal.ParseFullMethod(fullMethod)
 	peerAttrs := peerAttr(peerAddress)
 
-	attrs := make([]attribute.KeyValue, 0, 1+len(mAttrs)+len(peerAttrs))
+	attrs := make([]attribute.KeyValue, 0, 1+len(methodAttrs)+len(peerAttrs))
 	attrs = append(attrs, RPCSystemGRPC)
-	attrs = append(attrs, mAttrs...)
+	attrs = append(attrs, methodAttrs...)
+	metricAttrs := attrs[:1+len(methodAttrs)]
 	attrs = append(attrs, peerAttrs...)
-	return name, attrs
+	return name, attrs, metricAttrs
 }
 
 // peerAttr returns attributes about the peer address.

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

@@ -56,7 +56,7 @@ func (s *metadataSupplier) Keys() []string {
 // requests.
 // Deprecated: Unnecessary public func.
 func Inject(ctx context.Context, md *metadata.MD, opts ...Option) {
-	c := newConfig(opts)
+	c := newConfig(opts, "")
 	c.Propagators.Inject(ctx, &metadataSupplier{
 		metadata: md,
 	})
@@ -78,7 +78,7 @@ func inject(ctx context.Context, propagators propagation.TextMapPropagator) cont
 // 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)
+	c := newConfig(opts, "")
 	ctx = c.Propagators.Extract(ctx, &metadataSupplier{
 		metadata: md,
 	})

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

@@ -17,13 +17,16 @@ package otelgrpc // import "go.opentelemetry.io/contrib/instrumentation/google.g
 import (
 	"context"
 	"sync/atomic"
+	"time"
 
 	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/attribute"
 	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/metric"
 	semconv "go.opentelemetry.io/otel/semconv/v1.17.0"
 	"go.opentelemetry.io/otel/trace"
 )
@@ -33,24 +36,32 @@ type gRPCContextKey struct{}
 type gRPCContext struct {
 	messagesReceived int64
 	messagesSent     int64
+	metricAttrs      []attribute.KeyValue
+}
+
+type serverHandler struct {
+	*config
 }
 
 // NewServerHandler creates a stats.Handler for gRPC server.
 func NewServerHandler(opts ...Option) stats.Handler {
 	h := &serverHandler{
-		config: newConfig(opts),
+		config: newConfig(opts, "server"),
 	}
 
-	h.tracer = h.config.TracerProvider.Tracer(
-		instrumentationName,
-		trace.WithInstrumentationVersion(SemVersion()),
-	)
 	return h
 }
 
-type serverHandler struct {
-	*config
-	tracer trace.Tracer
+// 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) {
 }
 
 // TagRPC can attach some information to the given context.
@@ -66,46 +77,30 @@ func (h *serverHandler) TagRPC(ctx context.Context, info *stats.RPCTagInfo) cont
 		trace.WithAttributes(attrs...),
 	)
 
-	gctx := gRPCContext{}
+	gctx := gRPCContext{
+		metricAttrs: attrs,
+	}
 	return context.WithValue(ctx, gRPCContextKey{}, &gctx)
 }
 
 // HandleRPC processes the RPC stats.
 func (h *serverHandler) HandleRPC(ctx context.Context, rs stats.RPCStats) {
-	handleRPC(ctx, rs)
+	h.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) {
+type clientHandler struct {
+	*config
 }
 
 // NewClientHandler creates a stats.Handler for gRPC client.
 func NewClientHandler(opts ...Option) stats.Handler {
 	h := &clientHandler{
-		config: newConfig(opts),
+		config: newConfig(opts, "client"),
 	}
 
-	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)
@@ -117,14 +112,16 @@ func (h *clientHandler) TagRPC(ctx context.Context, info *stats.RPCTagInfo) cont
 		trace.WithAttributes(attrs...),
 	)
 
-	gctx := gRPCContext{}
+	gctx := gRPCContext{
+		metricAttrs: attrs,
+	}
 
 	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)
+	h.handleRPC(ctx, rs)
 }
 
 // TagConn can attach some information to the given context.
@@ -140,48 +137,102 @@ func (h *clientHandler) HandleConn(context.Context, stats.ConnStats) {
 	// no-op
 }
 
-func handleRPC(ctx context.Context, rs stats.RPCStats) {
+func (c *config) handleRPC(ctx context.Context, rs stats.RPCStats) {
 	span := trace.SpanFromContext(ctx)
 	gctx, _ := ctx.Value(gRPCContextKey{}).(*gRPCContext)
 	var messageId int64
+	metricAttrs := make([]attribute.KeyValue, 0, len(gctx.metricAttrs)+1)
+	metricAttrs = append(metricAttrs, gctx.metricAttrs...)
+	wctx := withoutCancel(ctx)
 
 	switch rs := rs.(type) {
 	case *stats.Begin:
 	case *stats.InPayload:
 		if gctx != nil {
 			messageId = atomic.AddInt64(&gctx.messagesReceived, 1)
+			c.rpcRequestSize.Record(wctx, int64(rs.Length), metric.WithAttributes(metricAttrs...))
+		}
+
+		if c.ReceivedEvent {
+			span.AddEvent("message",
+				trace.WithAttributes(
+					semconv.MessageTypeReceived,
+					semconv.MessageIDKey.Int64(messageId),
+					semconv.MessageCompressedSizeKey.Int(rs.CompressedLength),
+					semconv.MessageUncompressedSizeKey.Int(rs.Length),
+				),
+			)
 		}
-		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)
+			c.rpcResponseSize.Record(wctx, int64(rs.Length), metric.WithAttributes(metricAttrs...))
 		}
 
-		span.AddEvent("message",
-			trace.WithAttributes(
-				semconv.MessageTypeSent,
-				semconv.MessageIDKey.Int64(messageId),
-				semconv.MessageCompressedSizeKey.Int(rs.CompressedLength),
-				semconv.MessageUncompressedSizeKey.Int(rs.Length),
-			),
-		)
+		if c.SentEvent {
+			span.AddEvent("message",
+				trace.WithAttributes(
+					semconv.MessageTypeSent,
+					semconv.MessageIDKey.Int64(messageId),
+					semconv.MessageCompressedSizeKey.Int(rs.CompressedLength),
+					semconv.MessageUncompressedSizeKey.Int(rs.Length),
+				),
+			)
+		}
+	case *stats.OutTrailer:
 	case *stats.End:
+		var rpcStatusAttr attribute.KeyValue
+
 		if rs.Error != nil {
 			s, _ := status.FromError(rs.Error)
 			span.SetStatus(codes.Error, s.Message())
-			span.SetAttributes(statusCodeAttr(s.Code()))
+			rpcStatusAttr = semconv.RPCGRPCStatusCodeKey.Int(int(s.Code()))
 		} else {
-			span.SetAttributes(statusCodeAttr(grpc_codes.OK))
+			rpcStatusAttr = semconv.RPCGRPCStatusCodeKey.Int(int(grpc_codes.OK))
 		}
+		span.SetAttributes(rpcStatusAttr)
 		span.End()
+
+		metricAttrs = append(metricAttrs, rpcStatusAttr)
+
+		// Use floating point division here for higher precision (instead of Millisecond method).
+		elapsedTime := float64(rs.EndTime.Sub(rs.BeginTime)) / float64(time.Millisecond)
+
+		c.rpcDuration.Record(wctx, elapsedTime, metric.WithAttributes(metricAttrs...))
+		c.rpcRequestsPerRPC.Record(wctx, atomic.LoadInt64(&gctx.messagesReceived), metric.WithAttributes(metricAttrs...))
+		c.rpcResponsesPerRPC.Record(wctx, atomic.LoadInt64(&gctx.messagesSent), metric.WithAttributes(metricAttrs...))
 	default:
 		return
 	}
 }
+
+func withoutCancel(parent context.Context) context.Context {
+	if parent == nil {
+		panic("cannot create context from nil parent")
+	}
+	return withoutCancelCtx{parent}
+}
+
+type withoutCancelCtx struct {
+	c context.Context
+}
+
+func (withoutCancelCtx) Deadline() (deadline time.Time, ok bool) {
+	return
+}
+
+func (withoutCancelCtx) Done() <-chan struct{} {
+	return nil
+}
+
+func (withoutCancelCtx) Err() error {
+	return nil
+}
+
+func (w withoutCancelCtx) Value(key any) any {
+	return w.c.Value(key)
+}
+
+func (w withoutCancelCtx) String() string {
+	return "withoutCancel"
+}

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

@@ -16,7 +16,7 @@ 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.45.0"
+	return "0.46.1"
 	// This string is updated by the pre_release.sh script during release
 }