xds/resolver: minor cleanup in the config selector implementation (#8609)

This PR injects the dependencies of `configSelector` at creation time,
instead of passing a reference to the `xdsResolver` and having the
former directly access fields from the latter.

RELEASE NOTES: none

Author: easwars

internal/xds/resolver/serviceconfig.go

@@ -151,8 +151,11 @@ func (cs *erroringConfigSelector) SelectConfig(iresolver.RPCInfo) (*iresolver.RP
 func (cs *erroringConfigSelector) stop() {}
 
 type configSelector struct {
-	r                *xdsResolver
-	xdsNodeID        string
+	channelID            uint64 // Static hash when hash policy is HashPolicyTypeChannelID
+	xdsNodeID            string // xDS node ID, for annotating errors.
+	sendNewServiceConfig func() // Function to send a new service config to gRPC.
+
+	// Configuration received from the xDS management server.
 	virtualHost      virtualHost
 	routes           []route
 	clusters         map[string]*clusterInfo
@@ -214,9 +217,7 @@ func (cs *configSelector) SelectConfig(rpcInfo iresolver.RPCInfo) (*iresolver.RP
 			if v := atomic.AddInt32(ref, -1); v == 0 {
 				// This entry will be removed from activeClusters when
 				// producing the service config for the empty update.
-				cs.r.serializer.TrySchedule(func(context.Context) {
-					cs.r.onClusterRefDownToZero()
-				})
+				cs.sendNewServiceConfig()
 			}
 		},
 		Interceptor: interceptor,
@@ -281,7 +282,7 @@ func (cs *configSelector) generateHash(rpcInfo iresolver.RPCInfo, hashPolicies [
 			generatedPolicyHash = true
 		case xdsresource.HashPolicyTypeChannelID:
 			// Use the static channel ID as the hash for this policy.
-			policyHash = cs.r.channelID
+			policyHash = cs.channelID
 			generatedHash = true
 			generatedPolicyHash = true
 		}
@@ -358,9 +359,7 @@ func (cs *configSelector) stop() {
 	// selector; we need another update to delete clusters from the config (if
 	// we don't have another update pending already).
 	if needUpdate {
-		cs.r.serializer.TrySchedule(func(context.Context) {
-			cs.r.onClusterRefDownToZero()
-		})
+		cs.sendNewServiceConfig()
 	}
 }
 

internal/xds/resolver/serviceconfig_test.go

@@ -29,7 +29,6 @@ import (
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/internal/grpcutil"
 	iresolver "google.golang.org/grpc/internal/resolver"
-	"google.golang.org/grpc/internal/testutils"
 	_ "google.golang.org/grpc/internal/xds/balancer/cdsbalancer" // To parse LB config
 	"google.golang.org/grpc/internal/xds/xdsclient/xdsresource"
 	"google.golang.org/grpc/metadata"
@@ -64,12 +63,8 @@ func (s) TestPruneActiveClusters(t *testing.T) {
 
 func (s) TestGenerateRequestHash(t *testing.T) {
 	const channelID = 12378921
-	cs := &configSelector{
-		r: &xdsResolver{
-			cc:        &testutils.ResolverClientConn{Logger: t},
-			channelID: channelID,
-		},
-	}
+	cs := &configSelector{channelID: channelID}
+
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	tests := []struct {

internal/xds/resolver/xds_resolver.go

@@ -330,8 +330,13 @@ func (r *xdsResolver) sendNewServiceConfig(cs stoppableConfigSelector) bool {
 // Only executed in the context of a serializer callback.
 func (r *xdsResolver) newConfigSelector() *configSelector {
 	cs := &configSelector{
-		r:         r,
+		channelID: r.channelID,
 		xdsNodeID: r.xdsClient.BootstrapConfig().Node().GetId(),
+		sendNewServiceConfig: func() {
+			r.serializer.TrySchedule(func(context.Context) {
+				r.sendNewServiceConfig(r.curConfigSelector)
+			})
+		},
 		virtualHost: virtualHost{
 			httpFilterConfigOverride: r.currentVirtualHost.HTTPFilterConfigOverride,
 			retryConfig:              r.currentVirtualHost.RetryConfig,
@@ -607,8 +612,3 @@ func (r *xdsResolver) onRouteConfigResourceError(name string, err error) {
 	}
 	r.onResourceError(err)
 }
-
-// Only executed in the context of a serializer callback.
-func (r *xdsResolver) onClusterRefDownToZero() {
-	r.sendNewServiceConfig(r.curConfigSelector)
-}