stats/otel: Add grpc.lb.backend_service label to wrr metrics (A89) (#8737)

mbissa · web-flow · commit f9d2bdb34edc · 2025-12-08T12:19:14.000+05:30
Address: https://github.com/grpc/proposal/blob/master/A89-backend-service-metric-label.md This PR makes available the `backend_service` (cluster name) label which is decided in clusterimpl (since we are pre-A75). It is added to WRR per-call metrics. RELEASE NOTES: * stats/otel: add backend service label to wrr metrics as part of A89
diff --git a/balancer/weightedroundrobin/balancer.go b/balancer/weightedroundrobin/balancer.go
@@ -62,7 +62,7 @@ var (
 		Description:    "EXPERIMENTAL. Number of scheduler updates in which there were not enough endpoints with valid weight, which caused the WRR policy to fall back to RR behavior.",
 		Unit:           "{update}",
 		Labels:         []string{"grpc.target"},
-		OptionalLabels: []string{"grpc.lb.locality"},
+		OptionalLabels: []string{"grpc.lb.locality", "grpc.lb.backend_service"},
 		Default:        false,
 	})
 
@@ -71,7 +71,7 @@ var (
 		Description:    "EXPERIMENTAL. Number of endpoints from each scheduler update that don't yet have usable weight information (i.e., either the load report has not yet been received, or it is within the blackout period).",
 		Unit:           "{endpoint}",
 		Labels:         []string{"grpc.target"},
-		OptionalLabels: []string{"grpc.lb.locality"},
+		OptionalLabels: []string{"grpc.lb.locality", "grpc.lb.backend_service"},
 		Default:        false,
 	})
 
@@ -80,15 +80,15 @@ var (
 		Description:    "EXPERIMENTAL. Number of endpoints from each scheduler update whose latest weight is older than the expiration period.",
 		Unit:           "{endpoint}",
 		Labels:         []string{"grpc.target"},
-		OptionalLabels: []string{"grpc.lb.locality"},
+		OptionalLabels: []string{"grpc.lb.locality", "grpc.lb.backend_service"},
 		Default:        false,
 	})
 	endpointWeightsMetric = estats.RegisterFloat64Histo(estats.MetricDescriptor{
 		Name:           "grpc.lb.wrr.endpoint_weights",
 		Description:    "EXPERIMENTAL. Weight of each endpoint, recorded on every scheduler update. Endpoints without usable weights will be recorded as weight 0.",
 		Unit:           "{endpoint}",
 		Labels:         []string{"grpc.target"},
-		OptionalLabels: []string{"grpc.lb.locality"},
+		OptionalLabels: []string{"grpc.lb.locality", "grpc.lb.backend_service"},
 		Default:        false,
 	})
 )
@@ -173,6 +173,7 @@ func (b *wrrBalancer) updateEndpointsLocked(endpoints []resolver.Endpoint) {
 				metricsRecorder: b.metricsRecorder,
 				target:          b.target,
 				locality:        b.locality,
+				clusterName:     b.clusterName,
 			}
 			for _, addr := range endpoint.Addresses {
 				b.addressWeights.Set(addr, ew)
@@ -211,6 +212,7 @@ type wrrBalancer struct {
 	mu               sync.Mutex
 	cfg              *lbConfig // active config
 	locality         string
+	clusterName      string
 	stopPicker       *grpcsync.Event
 	addressWeights   *resolver.AddressMapV2[*endpointWeight]
 	endpointToWeight *resolver.EndpointMap[*endpointWeight]
@@ -231,6 +233,7 @@ func (b *wrrBalancer) UpdateClientConnState(ccs balancer.ClientConnState) error
 	b.mu.Lock()
 	b.cfg = cfg
 	b.locality = weightedtarget.LocalityFromResolverState(ccs.ResolverState)
+	b.clusterName = backendServiceFromState(ccs.ResolverState)
 	b.updateEndpointsLocked(ccs.ResolverState.Endpoints)
 	b.mu.Unlock()
 
@@ -288,6 +291,7 @@ func (b *wrrBalancer) UpdateState(state balancer.State) {
 		metricsRecorder: b.metricsRecorder,
 		locality:        b.locality,
 		target:          b.target,
+		clusterName:     b.clusterName,
 	}
 
 	b.stopPicker = grpcsync.NewEvent()
@@ -420,6 +424,7 @@ type picker struct {
 	// The following fields are immutable.
 	target          string
 	locality        string
+	clusterName     string
 	metricsRecorder estats.MetricsRecorder
 }
 
@@ -499,6 +504,7 @@ type endpointWeight struct {
 	target          string
 	metricsRecorder estats.MetricsRecorder
 	locality        string
+	clusterName     string
 
 	// The following fields are only accessed on calls into the LB policy, and
 	// do not need a mutex.
@@ -602,14 +608,14 @@ func (w *endpointWeight) weight(now time.Time, weightExpirationPeriod, blackoutP
 
 	if recordMetrics {
 		defer func() {
-			endpointWeightsMetric.Record(w.metricsRecorder, weight, w.target, w.locality)
+			endpointWeightsMetric.Record(w.metricsRecorder, weight, w.target, w.locality, w.clusterName)
 		}()
 	}
 
 	// The endpoint has not received a load report (i.e. just turned READY with
 	// no load report).
 	if w.lastUpdated.Equal(time.Time{}) {
-		endpointWeightNotYetUsableMetric.Record(w.metricsRecorder, 1, w.target, w.locality)
+		endpointWeightNotYetUsableMetric.Record(w.metricsRecorder, 1, w.target, w.locality, w.clusterName)
 		return 0
 	}
 
@@ -618,7 +624,7 @@ func (w *endpointWeight) weight(now time.Time, weightExpirationPeriod, blackoutP
 	// start getting data again in the future, and return 0.
 	if now.Sub(w.lastUpdated) >= weightExpirationPeriod {
 		if recordMetrics {
-			endpointWeightStaleMetric.Record(w.metricsRecorder, 1, w.target, w.locality)
+			endpointWeightStaleMetric.Record(w.metricsRecorder, 1, w.target, w.locality, w.clusterName)
 		}
 		w.nonEmptySince = time.Time{}
 		return 0
@@ -627,10 +633,27 @@ func (w *endpointWeight) weight(now time.Time, weightExpirationPeriod, blackoutP
 	// If we don't have at least blackoutPeriod worth of data, return 0.
 	if blackoutPeriod != 0 && (w.nonEmptySince.Equal(time.Time{}) || now.Sub(w.nonEmptySince) < blackoutPeriod) {
 		if recordMetrics {
-			endpointWeightNotYetUsableMetric.Record(w.metricsRecorder, 1, w.target, w.locality)
+			endpointWeightNotYetUsableMetric.Record(w.metricsRecorder, 1, w.target, w.locality, w.clusterName)
 		}
 		return 0
 	}
 
 	return w.weightVal
 }
+
+type backendServiceKey struct{}
+
+// SetBackendService stores the backendService on the resolver state so
+// that it can be used later as a label in wrr metrics.
+func SetBackendService(state resolver.State, backendService string) resolver.State {
+	state.Attributes = state.Attributes.WithValue(backendServiceKey{}, backendService)
+	return state
+}
+
+// getBackendServiceFromState retrieves the cluster name stored as an attribute
+// in the resolver state.
+func backendServiceFromState(state resolver.State) string {
+	v := state.Attributes.Value(backendServiceKey{})
+	name, _ := v.(string)
+	return name
+}
diff --git a/balancer/weightedroundrobin/scheduler.go b/balancer/weightedroundrobin/scheduler.go
@@ -39,7 +39,7 @@ func (p *picker) newScheduler(recordMetrics bool) scheduler {
 	}
 	if n == 1 {
 		if recordMetrics {
-			rrFallbackMetric.Record(p.metricsRecorder, 1, p.target, p.locality)
+			rrFallbackMetric.Record(p.metricsRecorder, 1, p.target, p.locality, p.clusterName)
 		}
 		return &rrScheduler{numSCs: 1, inc: p.inc}
 	}
@@ -58,7 +58,7 @@ func (p *picker) newScheduler(recordMetrics bool) scheduler {
 
 	if numZero >= n-1 {
 		if recordMetrics {
-			rrFallbackMetric.Record(p.metricsRecorder, 1, p.target, p.locality)
+			rrFallbackMetric.Record(p.metricsRecorder, 1, p.target, p.locality, p.clusterName)
 		}
 		return &rrScheduler{numSCs: uint32(n), inc: p.inc}
 	}
diff --git a/internal/xds/balancer/clusterimpl/clusterimpl.go b/internal/xds/balancer/clusterimpl/clusterimpl.go
@@ -33,6 +33,7 @@ import (
 	"time"
 
 	"google.golang.org/grpc/balancer"
+	"google.golang.org/grpc/balancer/weightedroundrobin"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/internal"
 	"google.golang.org/grpc/internal/balancer/gracefulswitch"
@@ -297,7 +298,7 @@ func (b *clusterImplBalancer) UpdateClientConnState(s balancer.ClientConnState)
 
 	// Addresses and sub-balancer config are sent to sub-balancer.
 	err = b.child.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState:  s.ResolverState,
+		ResolverState:  weightedroundrobin.SetBackendService(s.ResolverState, b.clusterName),
 		BalancerConfig: parsedCfg,
 	})
 
diff --git a/stats/opentelemetry/csm/observability_test.go b/stats/opentelemetry/csm/observability_test.go
@@ -434,7 +434,8 @@ func unaryInterceptorAttachXDSLabels(ctx context.Context, method string, req, re
 			"csm.service_name":           "service_name_val",
 			"csm.service_namespace_name": "service_namespace_val",
 
-			"grpc.lb.locality": "grpc.lb.locality_val",
+			"grpc.lb.locality":        "grpc.lb.locality_val",
+			"grpc.lb.backend_service": "grpc.lb.backend_service_val",
 		},
 	})
 
@@ -469,7 +470,7 @@ func (s) TestXDSLabels(t *testing.T) {
 		MetricsOptions: opentelemetry.MetricsOptions{
 			MeterProvider:  provider,
 			Metrics:        opentelemetry.DefaultMetrics(),
-			OptionalLabels: []string{"csm.service_name", "csm.service_namespace_name", "grpc.lb.locality"},
+			OptionalLabels: []string{"csm.service_name", "csm.service_namespace_name", "grpc.lb.locality", "grpc.lb.backend_service"},
 		},
 	}, po), grpc.WithUnaryInterceptor(unaryInterceptorAttachXDSLabels)}
 	if err := ss.Start(nil, dopts...); err != nil {
@@ -498,6 +499,7 @@ func (s) TestXDSLabels(t *testing.T) {
 	serviceNameAttr := attribute.String("csm.service_name", "service_name_val")
 	serviceNamespaceAttr := attribute.String("csm.service_namespace_name", "service_namespace_val")
 	localityAttr := attribute.String("grpc.lb.locality", "grpc.lb.locality_val")
+	backendServiceAttr := attribute.String("grpc.lb.backend_service", "grpc.lb.backend_service_val")
 	meshIDAttr := attribute.String("csm.mesh_id", "unknown")
 	workloadCanonicalServiceAttr := attribute.String("csm.workload_canonical_service", "unknown")
 	remoteWorkloadTypeAttr := attribute.String("csm.remote_workload_type", "unknown")
@@ -510,6 +512,7 @@ func (s) TestXDSLabels(t *testing.T) {
 		serviceNameAttr,
 		serviceNamespaceAttr,
 		localityAttr,
+		backendServiceAttr,
 		meshIDAttr,
 		workloadCanonicalServiceAttr,
 		remoteWorkloadTypeAttr,
diff --git a/stats/opentelemetry/e2e_test.go b/stats/opentelemetry/e2e_test.go
@@ -673,7 +673,7 @@ func (s) TestWRRMetrics(t *testing.T) {
 	mo := opentelemetry.MetricsOptions{
 		MeterProvider:  provider,
 		Metrics:        opentelemetry.DefaultMetrics().Add("grpc.lb.wrr.rr_fallback", "grpc.lb.wrr.endpoint_weight_not_yet_usable", "grpc.lb.wrr.endpoint_weight_stale", "grpc.lb.wrr.endpoint_weights"),
-		OptionalLabels: []string{"grpc.lb.locality"},
+		OptionalLabels: []string{"grpc.lb.locality", "grpc.lb.backend_service"},
 	}
 
 	target := fmt.Sprintf("xds:///%s", serviceName)
@@ -699,6 +699,7 @@ func (s) TestWRRMetrics(t *testing.T) {
 
 	targetAttr := attribute.String("grpc.target", target)
 	localityAttr := attribute.String("grpc.lb.locality", `{region="region-1", zone="zone-1", sub_zone="subzone-1"}`)
+	backendServiceAttr := attribute.String("grpc.lb.backend_service", clusterName)
 
 	wantMetrics := []metricdata.Metrics{
 		{
@@ -708,7 +709,7 @@ func (s) TestWRRMetrics(t *testing.T) {
 			Data: metricdata.Sum[int64]{
 				DataPoints: []metricdata.DataPoint[int64]{
 					{
-						Attributes: attribute.NewSet(targetAttr, localityAttr),
+						Attributes: attribute.NewSet(targetAttr, localityAttr, backendServiceAttr),
 						Value:      1, // value ignored
 					},
 				},
@@ -724,7 +725,7 @@ func (s) TestWRRMetrics(t *testing.T) {
 			Data: metricdata.Sum[int64]{
 				DataPoints: []metricdata.DataPoint[int64]{
 					{
-						Attributes: attribute.NewSet(targetAttr, localityAttr),
+						Attributes: attribute.NewSet(targetAttr, localityAttr, backendServiceAttr),
 						Value:      1, // value ignored
 					},
 				},
@@ -739,7 +740,7 @@ func (s) TestWRRMetrics(t *testing.T) {
 			Data: metricdata.Histogram[float64]{
 				DataPoints: []metricdata.HistogramDataPoint[float64]{
 					{
-						Attributes: attribute.NewSet(targetAttr, localityAttr),
+						Attributes: attribute.NewSet(targetAttr, localityAttr, backendServiceAttr),
 					},
 				},
 				Temporality: metricdata.CumulativeTemporality,
@@ -761,7 +762,7 @@ func (s) TestWRRMetrics(t *testing.T) {
 		Data: metricdata.Sum[int64]{
 			DataPoints: []metricdata.DataPoint[int64]{
 				{
-					Attributes: attribute.NewSet(targetAttr, localityAttr),
+					Attributes: attribute.NewSet(targetAttr, localityAttr, backendServiceAttr),
 					Value:      1, // value ignored
 				},
 			},

Original file line number	Diff line number	Diff line change
`@@ -39,7 +39,7 @@ func (p *picker) newScheduler(recordMetrics bool) scheduler {`
`39`	`39`	`}`
`40`	`40`	`if n == 1 {`
`41`	`41`	`if recordMetrics {`
`42`		`- rrFallbackMetric.Record(p.metricsRecorder, 1, p.target, p.locality)`
	`42`	`+ rrFallbackMetric.Record(p.metricsRecorder, 1, p.target, p.locality, p.clusterName)`
`43`	`43`	`}`
`44`	`44`	`return &rrScheduler{numSCs: 1, inc: p.inc}`
`45`	`45`	`}`
`@@ -58,7 +58,7 @@ func (p *picker) newScheduler(recordMetrics bool) scheduler {`
`58`	`58`
`59`	`59`	`if numZero >= n-1 {`
`60`	`60`	`if recordMetrics {`
`61`		`- rrFallbackMetric.Record(p.metricsRecorder, 1, p.target, p.locality)`
	`61`	`+ rrFallbackMetric.Record(p.metricsRecorder, 1, p.target, p.locality, p.clusterName)`
`62`	`62`	`}`
`63`	`63`	`return &rrScheduler{numSCs: uint32(n), inc: p.inc}`
`64`	`64`	`}`