stats/opentelemetry: Add support for optional label grpc.lb.backend_service in per-call metrics

internal/xds/balancer/clusterimpl/clusterimpl.go

@@ -163,6 +163,7 @@ func (b *clusterImplBalancer) newPickerLocked() *picker {
 		counter:         b.requestCounter,
 		countMax:        b.requestCountMax,
 		telemetryLabels: b.telemetryLabels,
+		clusterName:     b.clusterName,
 	}
 }
 
@@ -414,6 +415,7 @@ func (b *clusterImplBalancer) getClusterName() string {
 // SubConn to the wrapper for this purpose.
 type scWrapper struct {
 	balancer.SubConn
+
 	// locality needs to be atomic because it can be updated while being read by
 	// the picker.
 	locality atomic.Pointer[clients.Locality]

internal/xds/balancer/clusterimpl/picker.go

@@ -87,6 +87,7 @@ type picker struct {
 	counter         *xdsclient.ClusterRequestsCounter
 	countMax        uint32
 	telemetryLabels map[string]string
+	clusterName     string
 }
 
 func telemetryLabels(ctx context.Context) map[string]string {
@@ -136,6 +137,7 @@ func (d *picker) Pick(info balancer.PickInfo) (balancer.PickResult, error) {
 	}
 
 	var lID clients.Locality
+
 	pr, err := d.s.Picker.Pick(info)
 	if scw, ok := pr.SubConn.(*scWrapper); ok {
 		// This OK check also covers the case err!=nil, because SubConn will be
@@ -156,6 +158,7 @@ func (d *picker) Pick(info balancer.PickInfo) (balancer.PickResult, error) {
 
 	if labels := telemetryLabels(info.Ctx); labels != nil {
 		labels["grpc.lb.locality"] = xdsinternal.LocalityString(lID)
+		labels["grpc.lb.backend_service"] = d.clusterName
 	}
 
 	if d.loadStore != nil {

stats/opentelemetry/client_metrics.go

@@ -174,7 +174,8 @@ func (h *clientMetricsHandler) TagRPC(ctx context.Context, info *stats.RPCTagInf
 			// executes on the callpath that this OpenTelemetry component
 			// currently supports.
 			TelemetryLabels: map[string]string{
-				"grpc.lb.locality": "",
+				"grpc.lb.locality":        "",
+				"grpc.lb.backend_service": "",
 			},
 		}
 		ctx = istats.SetLabels(ctx, labels)

test/xds/xds_telemetry_labels_test.go

@@ -44,7 +44,8 @@ const serviceNamespaceKey = "service_namespace"
 const serviceNamespaceKeyCSM = "csm.service_namespace_name"
 const serviceNameValue = "grpc-service"
 const serviceNamespaceValue = "grpc-service-namespace"
-
+const backendServiceKey = "grpc.lb.backend_service"
+const backendServiceValue = "cluster-my-service-client-side-xds"
 const localityKey = "grpc.lb.locality"
 const localityValue = `{region="region-1", zone="zone-1", sub_zone="subzone-1"}`
 
@@ -124,23 +125,23 @@ func (fsh *fakeStatsHandler) TagRPC(ctx context.Context, _ *stats.RPCTagInfo) co
 
 func (fsh *fakeStatsHandler) HandleRPC(_ context.Context, rs stats.RPCStats) {
 	switch rs.(type) {
-	// stats.Begin won't get Telemetry Labels because happens after picker
-	// picks.
-
-	// These three stats callouts trigger all metrics for OpenTelemetry that
-	// aren't started. All of these should have access to the desired telemetry
+	// stats.Begin is called before the picker runs, so it won't have telemetry
 	// labels.
+	// The following three stats callouts trigger OpenTelemetry metrics and are
+	// guaranteed to run after the picker has selected a subchannel. Therefore,
+	// they should have access to the desired telemetry labels.
 	case *stats.OutPayload, *stats.InPayload, *stats.End:
 		want := map[string]string{
 			serviceNameKeyCSM:      serviceNameValue,
 			serviceNamespaceKeyCSM: serviceNamespaceValue,
 			localityKey:            localityValue,
+			backendServiceKey:      backendServiceValue,
 		}
 		if diff := cmp.Diff(fsh.labels.TelemetryLabels, want); diff != "" {
 			fsh.t.Fatalf("fsh.labels.TelemetryLabels (-got +want): %v", diff)
 		}
 	default:
 		// Nothing to assert for the other stats.Handler callouts.
-	}
 
+	}
 }