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

Code before:
		method = "other"
	} else {
		isRegisteredMethod := internal.IsRegisteredMethod.(func(*grpc.Server, string) bool)
		if !isRegisteredMethod(server, method) {
			method = "other"
		}
	}

	ai := &attemptInfo{
		startTime: time.Now(),
		method:    removeLeadingSlash(method),
	}
	return setRPCInfo(ctx, &rpcInfo{
		ai: ai,
	})
}

func (h *serverTracingHandler) TagRPC(ctx context.Context, _ *stats.RPCTagInfo) context.Context {
	ri := getRPCInfo(ctx)
	if ri == nil {
		logger.Error("ctx passed into server side stats handler metrics event handling has no server attempt data present")
		return ctx
	}
	ai := ri.ai

	ctx, _ = h.traceTagRPC(ctx, ai)
	return ctx
}

// HandleRPC implements per RPC tracing and stats implementation for metrics.
func (h *serverStatsHandler) HandleRPC(ctx context.Context, rs stats.RPCStats) {
	ri := getRPCInfo(ctx)
	if ri == nil {
		logger.Error("ctx passed into server side stats handler metrics event handling has no server call data present")
		return
	}
	h.processRPCData(ctx, rs, ri.ai)
}

// HandleRPC implements per RPC tracing and stats implementation for tracing.
func (h *serverTracingHandler) HandleRPC(ctx context.Context, rs stats.RPCStats) {
	ri := getRPCInfo(ctx)
	if ri == nil {
		logger.Error("ctx passed into server side tracing stats handler has no server call data present")
		return
	}
	populateSpan(rs, ri.ai)
}

func (h *serverStatsHandler) processRPCData(ctx context.Context, s stats.RPCStats, ai *attemptInfo) {
	switch st := s.(type) {

Review comment (suggestion):
we shouldn't ignore the attempt info because that contains the trace span. We should set it again in the context using `setRPCInfo`

Code after:
			}
			ai.pluginOptionLabels = labels
		}
		attrs := otelmetric.WithAttributeSet(otelattribute.NewSet(
			otelattribute.String("grpc.method", ai.method),
		))
		h.serverMetrics.callStarted.Add(ctx, 1, attrs)
	case *stats.OutPayload:
		atomic.AddInt64(&ai.sentCompressedBytes, int64(st.CompressedLength))
	case *stats.InPayload:
		atomic.AddInt64(&ai.recvCompressedBytes, int64(st.CompressedLength))
	case *stats.End:
		h.processRPCEnd(ctx, ai, st)
	default:
	}
}

func (h *serverMetricsHandler) processRPCEnd(ctx context.Context, ai *attemptInfo, e *stats.End) {
	latency := float64(time.Since(ai.startTime)) / float64(time.Second)
	st := "OK"
	if e.Error != nil {
		s, _ := status.FromError(e.Error)
		st = canonicalString(s.Code())
	}
	attributes := []otelattribute.KeyValue{
		otelattribute.String("grpc.method", ai.method),
		otelattribute.String("grpc.status", st),
	}
	for k, v := range ai.pluginOptionLabels {
		attributes = append(attributes, otelattribute.String(k, v))
	}

	// Allocate vararg slice once.
	opts := []otelmetric.RecordOption{otelmetric.WithAttributeSet(otelattribute.NewSet(attributes...))}
	h.serverMetrics.callDuration.Record(ctx, latency, opts...)
	h.serverMetrics.callSentTotalCompressedMessageSize.Record(ctx, atomic.LoadInt64(&ai.sentCompressedBytes), opts...)
	h.serverMetrics.callRcvdTotalCompressedMessageSize.Record(ctx, atomic.LoadInt64(&ai.recvCompressedBytes), opts...)
}

const (
	// ServerCallStartedMetricName is the number of server calls started.
	ServerCallStartedMetricName string = "grpc.server.call.started"
	// ServerCallSentCompressedTotalMessageSizeMetricName is the compressed
	// message bytes sent per server call.
	ServerCallSentCompressedTotalMessageSizeMetricName string = "grpc.server.call.sent_total_compressed_message_size"
	// ServerCallRcvdCompressedTotalMessageSizeMetricName is the compressed
	// message bytes received per server call.
	ServerCallRcvdCompressedTotalMessageSizeMetricName string = "grpc.server.call.rcvd_total_compressed_message_size"
	// ServerCallDurationMetricName is the end-to-end time taken to complete a
	// call from server transport's perspective.
	ServerCallDurationMetricName string = "grpc.server.call.duration"