a89 wrr changes adding backend service to per call metrics

    a94 subchannel metrics with labels

Author: mbissa

balancer/pickfirst/metrics_test.go

@@ -102,11 +102,31 @@ func (s) TestPickFirstMetrics(t *testing.T) {
 		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.lb.pick_first.disconnections", got, 0)
 	}
 
+	//Checking for subchannel metrics as well
+	if got, _ := tmr.Metric("grpc.subchannel.connection_attempts_succeeded"); got != 1 {
+		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.subchannel.connection_attempts_succeeded", got, 1)
+	}
+	if got, _ := tmr.Metric("grpc.subchannel.connection_attempts_failed"); got != 0 {
+		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.subchannel.connection_attempts_failed", got, 0)
+	}
+	if got, _ := tmr.Metric("grpc.subchannel.disconnections"); got != 0 {
+		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.subchannel.disconnections", got, 0)
+	}
+	if got, _ := tmr.Metric("grpc.subchannel.open_connections"); got != 1 {
+		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.subchannel.open_connections", got, 1)
+	}
+
 	ss.Stop()
 	testutils.AwaitState(ctx, t, cc, connectivity.Idle)
 	if got, _ := tmr.Metric("grpc.lb.pick_first.disconnections"); got != 1 {
 		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.lb.pick_first.disconnections", got, 1)
 	}
+	if got, _ := tmr.Metric("grpc.subchannel.disconnections"); got != 1 {
+		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.subchannel.disconnections", got, 1)
+	}
+	if got, _ := tmr.Metric("grpc.subchannel.open_connections"); got != -1 {
+		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.subchannel.open_connections", got, -1)
+	}
 }
 
 // TestPickFirstMetricsFailure tests the connection attempts failed metric. It

balancer/pickfirst/pickfirst_ext_test.go

@@ -1946,6 +1946,16 @@ func (s) TestPickFirstLeaf_HappyEyeballs_TF_AfterEndOfList(t *testing.T) {
 	if got, _ := tmr.Metric("grpc.lb.pick_first.disconnections"); got != 0 {
 		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.lb.pick_first.disconnections", got, 0)
 	}
+
+	if got, _ := tmr.Metric("grpc.subchannel.connection_attempts_succeeded"); got != 0 {
+		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.subchannel.connection_attempts_succeeded", got, 0)
+	}
+	if got, _ := tmr.Metric("grpc.subchannel.connection_attempts_failed"); got != 1 {
+		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.subchannel.connection_attempts_failed", got, 1)
+	}
+	if got, _ := tmr.Metric("grpc.subchannel.disconnections"); got != 0 {
+		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.subchannel.disconnections", got, 0)
+	}
 }
 
 // Test verifies that pickfirst attempts to connect to the second backend once
@@ -2006,6 +2016,16 @@ func (s) TestPickFirstLeaf_HappyEyeballs_TriggerConnectionDelay(t *testing.T) {
 	if got, _ := tmr.Metric("grpc.lb.pick_first.disconnections"); got != 0 {
 		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.lb.pick_first.disconnections", got, 0)
 	}
+
+	if got, _ := tmr.Metric("grpc.subchannel.connection_attempts_succeeded"); got != 1 {
+		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.subchannel.connection_attempts_succeeded", got, 1)
+	}
+	if got, _ := tmr.Metric("grpc.subchannel.disconnections"); got != 0 {
+		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.subchannel.disconnections", got, 0)
+	}
+	if got, _ := tmr.Metric("grpc.subchannel.open_connections"); got != 1 {
+		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.subchannel.open_connections", got, 1)
+	}
 }
 
 // Test tests the pickfirst balancer by causing a SubConn to fail and then
@@ -2057,6 +2077,9 @@ func (s) TestPickFirstLeaf_HappyEyeballs_TF_ThenTimerFires(t *testing.T) {
 	if got, _ := tmr.Metric("grpc.lb.pick_first.connection_attempts_failed"); got != 1 {
 		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.lb.pick_first.connection_attempts_failed", got, 1)
 	}
+	if got, _ := tmr.Metric("grpc.subchannel.connection_attempts_failed"); got != 1 {
+		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.subchannel.connection_attempts_failed", got, 1)
+	}
 	if holds[2].IsStarted() != false {
 		t.Fatalf("Server %d with address %q contacted unexpectedly", 2, addrs[2])
 	}
@@ -2080,6 +2103,13 @@ func (s) TestPickFirstLeaf_HappyEyeballs_TF_ThenTimerFires(t *testing.T) {
 	if got, _ := tmr.Metric("grpc.lb.pick_first.disconnections"); got != 0 {
 		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.lb.pick_first.disconnections", got, 0)
 	}
+
+	if got, _ := tmr.Metric("grpc.subchannel.connection_attempts_succeeded"); got != 1 {
+		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.subchannel.connection_attempts_succeeded", got, 1)
+	}
+	if got, _ := tmr.Metric("grpc.subchannel.disconnections"); got != 0 {
+		t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.subchannel.disconnections", got, 0)
+	}
 }
 
 func (s) TestPickFirstLeaf_InterleavingIPV4Preferred(t *testing.T) {

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,
+				cluster:         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,11 @@ func (b *wrrBalancer) UpdateClientConnState(ccs balancer.ClientConnState) error
 	b.mu.Lock()
 	b.cfg = cfg
 	b.locality = weightedtarget.LocalityFromResolverState(ccs.ResolverState)
+	if cluster, ok := resolver.GetBackendServiceFromState(ccs.ResolverState); !ok {
+		b.logger.Infof("Backend service name not found in resolver state attributes.")
+	} else {
+		b.clusterName = cluster
+	}
 	b.updateEndpointsLocked(ccs.ResolverState.Endpoints)
 	b.mu.Unlock()
 
@@ -288,6 +295,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 +428,7 @@ type picker struct {
 	// The following fields are immutable.
 	target          string
 	locality        string
+	clusterName     string
 	metricsRecorder estats.MetricsRecorder
 }
 
@@ -499,6 +508,7 @@ type endpointWeight struct {
 	target          string
 	metricsRecorder estats.MetricsRecorder
 	locality        string
+	cluster         string
 
 	// The following fields are only accessed on calls into the LB policy, and
 	// do not need a mutex.
@@ -602,14 +612,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.cluster)
 		}()
 	}
 
 	// 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.cluster)
 		return 0
 	}
 
@@ -618,7 +628,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.cluster)
 		}
 		w.nonEmptySince = time.Time{}
 		return 0
@@ -627,7 +637,7 @@ 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.cluster)
 		}
 		return 0
 	}

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}
 	}

clientconn.go

@@ -35,6 +35,8 @@ import (
 	"google.golang.org/grpc/balancer/pickfirst"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/connectivity"
+	"google.golang.org/grpc/credentials"
+	expstats "google.golang.org/grpc/experimental/stats"
 	"google.golang.org/grpc/internal"
 	"google.golang.org/grpc/internal/channelz"
 	"google.golang.org/grpc/internal/grpcsync"
@@ -98,6 +100,41 @@ var (
 	errTransportCredentialsMissing = errors.New("grpc: the credentials require transport level security (use grpc.WithTransportCredentials() to set)")
 )
 
+var (
+	disconnectionsMetric = expstats.RegisterInt64Count(expstats.MetricDescriptor{
+		Name:           "grpc.subchannel.disconnections",
+		Description:    "EXPERIMENTAL. Number of times the selected subchannel becomes disconnected.",
+		Unit:           "{disconnection}",
+		Labels:         []string{"grpc.target"},
+		OptionalLabels: []string{"grpc.lb.backend_service", "grpc.lb.locality", "grpc.disconnect_error"},
+		Default:        false,
+	})
+	connectionAttemptsSucceededMetric = expstats.RegisterInt64Count(expstats.MetricDescriptor{
+		Name:           "grpc.subchannel.connection_attempts_succeeded",
+		Description:    "EXPERIMENTAL. Number of successful connection attempts.",
+		Unit:           "{attempt}",
+		Labels:         []string{"grpc.target"},
+		OptionalLabels: []string{"grpc.lb.backend_service", "grpc.lb.locality"},
+		Default:        false,
+	})
+	connectionAttemptsFailedMetric = expstats.RegisterInt64Count(expstats.MetricDescriptor{
+		Name:           "grpc.subchannel.connection_attempts_failed",
+		Description:    "EXPERIMENTAL. Number of failed connection attempts.",
+		Unit:           "{attempt}",
+		Labels:         []string{"grpc.target"},
+		OptionalLabels: []string{"grpc.lb.backend_service", "grpc.lb.locality"},
+		Default:        false,
+	})
+	openConnectionsMetric = expstats.RegisterInt64UpDownCount(expstats.MetricDescriptor{
+		Name:           "grpc.subchannel.open_connections",
+		Description:    "EXPERIMENTAL. Number of open connections.",
+		Unit:           "{attempt}",
+		Labels:         []string{"grpc.target"},
+		OptionalLabels: []string{"grpc.lb.backend_service", "grpc.security_level", "grpc.lb.locality"},
+		Default:        false,
+	})
+)
+
 const (
 	defaultClientMaxReceiveMessageSize = 1024 * 1024 * 4
 	defaultClientMaxSendMessageSize    = math.MaxInt32
@@ -1223,6 +1260,11 @@ func (ac *addrConn) updateConnectivityState(s connectivity.State, lastErr error)
 	if ac.state == s {
 		return
 	}
+	locality, backendService := fetchLabels(ac)
+	if ac.state == connectivity.Ready || (ac.state == connectivity.Connecting && s == connectivity.Idle) {
+		disconnectionsMetric.Record(ac.cc.metricsRecorderList, 1, ac.cc.target, backendService, locality, "unknown")
+		openConnectionsMetric.Record(ac.cc.metricsRecorderList, -1, ac.cc.target, backendService, ac.securityLevel(), locality)
+	}
 	ac.state = s
 	ac.channelz.ChannelMetrics.State.Store(&s)
 	if lastErr == nil {
@@ -1276,10 +1318,13 @@ func (ac *addrConn) resetTransportAndUnlock() {
 	// https://github.com/grpc/grpc/blob/master/doc/connection-backoff.md#proposed-backoff-algorithm
 	connectDeadline := time.Now().Add(dialDuration)
 
+	locality, backendService := fetchLabels(ac)
+
 	ac.updateConnectivityState(connectivity.Connecting, nil)
 	ac.mu.Unlock()
 
 	if err := ac.tryAllAddrs(acCtx, addrs, connectDeadline); err != nil {
+		connectionAttemptsFailedMetric.Record(ac.cc.metricsRecorderList, 1, ac.cc.target, backendService, locality)
 		// TODO: #7534 - Move re-resolution requests into the pick_first LB policy
 		// to ensure one resolution request per pass instead of per subconn failure.
 		ac.cc.resolveNow(resolver.ResolveNowOptions{})
@@ -1319,10 +1364,43 @@ func (ac *addrConn) resetTransportAndUnlock() {
 	}
 	// Success; reset backoff.
 	ac.mu.Lock()
+	connectionAttemptsSucceededMetric.Record(ac.cc.metricsRecorderList, 1, ac.cc.target, backendService, locality)
+	openConnectionsMetric.Record(ac.cc.metricsRecorderList, 1, ac.cc.target, backendService, ac.securityLevel(), locality)
 	ac.backoffIdx = 0
 	ac.mu.Unlock()
 }
 
+func fetchLabels(ac *addrConn) (string, string) {
+	var locality, backendService string
+	labelsFromAddress, ok := internal.AddressToTelemetryLabels.(func(resolver.Address) map[string]string)
+	if len(ac.addrs) > 0 && internal.AddressToTelemetryLabels != nil && ok {
+		if ok {
+			labels := labelsFromAddress(ac.addrs[0])
+			locality = labels["grpc.lb.locality"]
+			backendService = labels["grpc.lb.backend_service"]
+		}
+	}
+	return locality, backendService
+}
+
+type securityLevelKey struct{}
+
+func (ac *addrConn) securityLevel() string {
+	var secLevel string
+	if ac.transport == nil {
+		secLevel, _ = ac.curAddr.Attributes.Value(securityLevelKey{}).(string)
+		return secLevel
+	}
+	authInfo := ac.transport.AuthInfo()
+	if ci, ok := authInfo.(interface {
+		GetCommonAuthInfo() credentials.CommonAuthInfo
+	}); ok {
+		secLevel = ci.GetCommonAuthInfo().SecurityLevel.String()
+		ac.curAddr.Attributes = ac.curAddr.Attributes.WithValue(securityLevelKey{}, secLevel)
+	}
+	return secLevel
+}
+
 // tryAllAddrs tries to create a connection to the addresses, and stop when at
 // the first successful one. It returns an error if no address was successfully
 // connected, or updates ac appropriately with the new transport.

internal/internal.go

@@ -244,6 +244,10 @@ var (
 	// When set, the function will be called before the stream enters
 	// the blocking state.
 	NewStreamWaitingForResolver = func() {}
+
+	// AddressToTelemetryLabels is an xDS-provided function to extract telemetry
+	// labels from a resolver.Address. Callers must assert its type before calling.
+	AddressToTelemetryLabels any // func(addr resolver.Address) map[string]string
 )
 
 // HealthChecker defines the signature of the client-side LB channel health