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

Code before:
				ctx = metadata.AppendToOutgoingContext(ctx, k, v)
			}
		}
	}

	startTime := time.Now()
	callback := func(err error) {
		h.perCallMetrics(ctx, err, startTime, ci)
	}
	opts = append([]grpc.CallOption{grpc.OnFinish(callback)}, opts...)
	return streamer(ctx, desc, cc, method, opts...)
}

// perCallMetrics records per call metrics.
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...)

Review comment (suggestion):
looks like we are setting callInfo again here. Since, tracingHandler is added after clientStatsHandler, the callInfo would have already been created. Similar to what we are doing with attempt info, we should try fetching here.

Code after:
	}
	opts = append([]grpc.CallOption{grpc.OnFinish(callback)}, opts...)
	return streamer(ctx, desc, cc, method, opts...)
}

// perCallMetrics records per call metrics for both unary and stream calls.
func (h *clientMetricsHandler) 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)
}

// 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": "",
			},
		}