xds: change cdsbalancer to use update from dependency manager (grpc#8907)

This PR is part of A74 changes. This PR changes following: 
1. Changes the CDS balancer to use the CDS and EDS/DNS update from
XDSConfig received by xds_resolver instead of starting its own CDS
watchers.
2. Remove cluster resolver completely
3. CDS balancer creates the priority config and creates priority
balancer as its child.
4. Remove serializer from CDS balancer since it was there to serializer
watcher updates and ClientConn updates, now that there are no watchers
and clientConn updates are already serialized , serializer is no longer
needed.
5. Update tests in cdsbalancer package to check the priority config
instead of cluster_resolver config.
6. xds_resolver tracks reference counts for weighted cluster and cluster
specifier plugin seperately. xds_resolver subscribes to the weighted
cluster in dependency manager when referenced for first time and
unsubscribes when references in xds_resolver go to zero.
7. Incase of LDS/RDS resource error , we now directly send empty service
config , instead of sending a complete service config with clusters that
RPCs are still referenced to.
8. Minor test fixes.

RELEASE NOTES: 
* xds: 
* Ambient errors for cluster resources are now logged exclusively in the
dependency manager and are no longer propagated to Load Balancing (LB)
policies.
* When re-resolution is requested, all clusters of type LOGICAL_DNS will
be re-resolved simultaneously, rather than only a single cluster.
* Upon receipt of a listener or route resource error, all in-flight RPCs
will now fail immediately.
* Any error encountered during the creation or update of a priority
configuration will now transition the channel to a TRANSIENT_FAILURE
state.

Author: eshitachandwani

internal/xds/balancer/balancer.go

@@ -25,7 +25,6 @@ import (
 	_ "google.golang.org/grpc/internal/xds/balancer/cdsbalancer"      // Register the CDS balancer
 	_ "google.golang.org/grpc/internal/xds/balancer/clusterimpl"      // Register the xds_cluster_impl balancer
 	_ "google.golang.org/grpc/internal/xds/balancer/clustermanager"   // Register the xds_cluster_manager balancer
-	_ "google.golang.org/grpc/internal/xds/balancer/clusterresolver"  // Register the xds_cluster_resolver balancer
 	_ "google.golang.org/grpc/internal/xds/balancer/outlierdetection" // Register the outlier_detection balancer
 	_ "google.golang.org/grpc/internal/xds/balancer/priority"         // Register the priority balancer
 )

internal/xds/balancer/cdsbalancer/aggregate_cluster_test.go

@@ -23,7 +23,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
-	"strings"
 	"testing"
 
 	"github.com/google/uuid"
@@ -552,14 +551,8 @@ func (s) TestSecurityConfigUpdate_GoodToFallback(t *testing.T) {
 // bootstrap file contents. Verifies that the connection between the client and
 // the server is secure. Subsequently, the cds LB policy receives a cluster
 // resource that is NACKed by the xDS client. Test verifies that the cds LB
-// policy continues to use the previous good configuration, but the error from
-// the xDS client is propagated to the child policy.
+// policy continues to use the previous good configuration.
 func (s) TestSecurityConfigUpdate_GoodToBad(t *testing.T) {
-	// Register a wrapped clusterresolver LB policy (child policy of the cds LB
-	// policy) for the duration of this test that makes the resolver error
-	// pushed to it available to the test.
-	_, resolverErrCh, _, _ := registerWrappedClusterResolverPolicy(t)
-
 	// Spin up an xDS management server.
 	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
 
@@ -615,16 +608,6 @@ func (s) TestSecurityConfigUpdate_GoodToBad(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	const wantNACKErr = "instance name \"unknown-certificate-provider-instance\" missing in bootstrap configuration"
-	select {
-	case err := <-resolverErrCh:
-		if !strings.Contains(err.Error(), wantNACKErr) {
-			t.Fatalf("Child policy got resolver error: %v, want err: %v", err, wantNACKErr)
-		}
-	case <-ctx.Done():
-		t.Fatal("Timeout when waiting for resolver error to be pushed to the child policy")
-	}
-
 	// Verify that a successful RPC can be made over a secure connection.
 	if _, err := client.EmptyCall(ctx, &testpb.Empty{}, grpc.WaitForReady(true)); err != nil {
 		t.Fatalf("EmptyCall() failed: %v", err)