Repository: grpc/grpc-go
File: stats/opentelemetry/client_metrics.go (Go)

Code before:
func (h *clientStatsHandler) perCallMetrics(ctx context.Context, err error, startTime time.Time, ci *callInfo) {
	callLatency := float64(time.Since(startTime)) / float64(time.Second)
	attrs := otelmetric.WithAttributeSet(otelattribute.NewSet(
		otelattribute.String("grpc.method", ci.method),
		otelattribute.String("grpc.target", ci.target),
		otelattribute.String("grpc.status", canonicalString(status.Code(err))),
	))
	h.clientMetrics.callDuration.Record(ctx, callLatency, attrs)
}

func (h *clientTracingHandler) unaryInterceptor(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
	ci := &callInfo{
		target: cc.CanonicalTarget(),
		method: determineMethod(method, opts...),
	}
	ctx = setCallInfo(ctx, ci)

	var span trace.Span
	ctx, span = h.createCallTraceSpan(ctx, method)
	err := invoker(ctx, method, req, reply, cc, opts...)
	h.perCallTraces(err, span)
	return err
}

func (h *clientTracingHandler) streamInterceptor(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
	ci := &callInfo{
		target: cc.CanonicalTarget(),
		method: determineMethod(method, opts...),
	}
	ctx = setCallInfo(ctx, ci)

	var span trace.Span
	ctx, span = h.createCallTraceSpan(ctx, method)
	callback := func(err error) {
		h.perCallTraces(err, span)
	}
	opts = append([]grpc.CallOption{grpc.OnFinish(callback)}, opts...)
	return streamer(ctx, desc, cc, method, opts...)
}

// perCallTraces records per call trace spans.
func (h *clientTracingHandler) perCallTraces(err error, ts trace.Span) {
	s := status.Convert(err)
	if s.Code() == grpccodes.OK {
		ts.SetStatus(otelcodes.Ok, s.Message())
	} else {
		ts.SetStatus(otelcodes.Error, s.Message())
	}
	ts.End()
}


Review comment (suggestion):
same comment about callInfo as `unaryInterceptor` of `tracingHandler`

Code after:
}

// TagConn exists to satisfy stats.Handler.
func (h *clientMetricsHandler) TagConn(ctx context.Context, _ *stats.ConnTagInfo) context.Context {
	return ctx
}

// HandleConn exists to satisfy stats.Handler.
func (h *clientMetricsHandler) HandleConn(context.Context, stats.ConnStats) {}

// getOrCreateRPCAttemptInfo retrieves or creates an rpc attemptInfo object
// and ensures it is set in the context along with the rpcInfo.
func getOrCreateRPCAttemptInfo(ctx context.Context) (context.Context, *attemptInfo) {
	ri := getRPCInfo(ctx)
	if ri != nil {
		return ctx, ri.ai
	}
	ri = &rpcInfo{ai: &attemptInfo{}}
	return setRPCInfo(ctx, ri), ri.ai
}

// TagRPC implements per RPC attempt context management for metrics.
func (h *clientMetricsHandler) TagRPC(ctx context.Context, info *stats.RPCTagInfo) context.Context {
	// Numerous stats handlers can be used for the same channel. The cluster
	// impl balancer which writes to this will only write once, thus have this
	// stats handler's per attempt scoped context point to the same optional
	// labels map if set.
	var labels *istats.Labels
	if labels = istats.GetLabels(ctx); labels == nil {
		labels = &istats.Labels{
			// The defaults for all the per call labels from a plugin that
			// executes on the callpath that this OpenTelemetry component
			// currently supports.
			TelemetryLabels: map[string]string{
				"grpc.lb.locality": "",
			},
		}
		ctx = istats.SetLabels(ctx, labels)
	}
	ctx, ai := getOrCreateRPCAttemptInfo(ctx)
	ai.startTime = time.Now()
	ai.xdsLabels = labels.TelemetryLabels
	ai.method = removeLeadingSlash(info.FullMethodName)

	return setRPCInfo(ctx, &rpcInfo{ai: ai})
}

// HandleRPC handles per RPC stats implementation.
func (h *clientMetricsHandler) HandleRPC(ctx context.Context, rs stats.RPCStats) {
	ri := getRPCInfo(ctx)
	if ri == nil {