xds: Add cluster endpoint watchers to depndency manager

internal/xds/resolver/watch_service_test.go

@@ -68,10 +68,14 @@ func (s) TestServiceWatch_ListenerPointsToNewRouteConfiguration(t *testing.T) {
 	verifyUpdateFromResolver(ctx, t, stateCh, wantServiceConfig(resources.Clusters[0].Name))
 
 	// Update the listener resource to point to a new route configuration name.
-	// Leave the old route configuration resource unchanged.
+	// The old route configuration resource is left unchanged to prevent a race:
+	// if it were removed immediately, the resolver might encounter a
+	// "resource-not-found" error for the old route before the listener update
+	// successfully transitions the client to the new route.
 	newTestRouteConfigName := defaultTestRouteConfigName + "-new"
 	resources.Listeners = []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, newTestRouteConfigName)}
-	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, resources.Listeners, resources.Routes)
+	resources.SkipValidation = true
+	mgmtServer.Update(ctx, resources)
 
 	// Verify that the new route configuration resource is requested.
 	waitForResourceNames(ctx, t, routeCfgCh, []string{newTestRouteConfigName})
@@ -88,9 +92,11 @@ func (s) TestServiceWatch_ListenerPointsToNewRouteConfiguration(t *testing.T) {
 			},
 		},
 	})
-	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, resources.Listeners, resources.Routes)
+	mgmtServer.Update(ctx, resources)
 
-	// Wait for no update from the resolver.
+	// Wait for no update from the resolver since the listener resource no
+	// longer points to the old route resource and new route resource hasn't
+	// been sent yet.
 	verifyNoUpdateFromResolver(ctx, t, stateCh)
 
 	// Update the management server with the new route configuration resource.

internal/xds/resolver/xds_resolver.go

@@ -230,8 +230,12 @@ type xdsResolver struct {
 	curConfigSelector stoppableConfigSelector
 }
 
-// ResolveNow is a no-op at this point.
-func (*xdsResolver) ResolveNow(resolver.ResolveNowOptions) {}
+// ResolveNow calls RequestDNSReresolution on the dependency manager.
+func (r *xdsResolver) ResolveNow(opts resolver.ResolveNowOptions) {
+	if r.dm != nil {
+		r.dm.RequestDNSReresolution(opts)
+	}
+}
 
 func (r *xdsResolver) Close() {
 	// Cancel the context passed to the serializer and wait for any scheduled

internal/xds/resolver/xds_resolver_test.go

@@ -436,14 +436,17 @@ func (s) TestResolverBadServiceUpdate_NACKedWithCache(t *testing.T) {
 
 	stateCh, _, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
 
-	// Configure good listener and route configuration resources on the
-	// management server.
-	listeners := []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)}
-	routes := []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(defaultTestRouteConfigName, defaultTestServiceName, defaultTestClusterName)}
-	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
+	// Configure all resources on the management server.
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		NodeID:     nodeID,
+		DialTarget: defaultTestServiceName,
+		Host:       "localhost",
+		Port:       8080,
+	})
+	mgmtServer.Update(ctx, resources)
 
 	// Expect a good update from the resolver.
-	cs := verifyUpdateFromResolver(ctx, t, stateCh, wantServiceConfig(defaultTestClusterName))
+	cs := verifyUpdateFromResolver(ctx, t, stateCh, wantServiceConfig(resources.Clusters[0].Name))
 
 	// "Make an RPC" by invoking the config selector.
 	_, err := cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"})
@@ -1001,7 +1004,9 @@ func (s) TestResolverDelayedOnCommitted(t *testing.T) {
 		Port:       defaultTestPort[0],
 		SecLevel:   e2e.SecurityLevelNone,
 	})
-	mgmtServer.Update(ctx, resources)
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
 
 	stateCh, _, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)