Repository: grpc/grpc-go

Issue:
Node ID should be available in RPC errors
1. Include node ID with all xdsclient errors possible

2. Audit all possible ways an xds-enabled gRPC channel can fail as a result of xdsclient problems, and ensure the xdsclient error is propagated to RPC errors


Discussion:
@dfawley: @easwars can you help scope this, please?@easwars: I suggest making the following changes:

- [x] The watch resource API call on the xDS client can fail for a bunch of reasons. We need to include the node ID in all these errors:
  - [x] A resource type implementation was already registered.
  - [x] Authority in the resource name is not found in the bootstrap configuration.
  - [x] Creation of transport to the management server fails.
- [x] A resource is NACKed:
  - [x] The `OnError` callback needs to include the xDS node ID in the returned error.
  - [x] The error is also stored in the cache to return to future watchers. This should include the xDS node ID as well.  
- [x] Reporting connectivity failures from xDS client to watchers. xDS node id should be part of these errors.
- [x] CDS lb policy needs to include the xDS node id in errors
  - [x] Requested resource is not found
  - [x] Sends a resolver error to child policy or puts the channel in TF
- [x] xDS resolver needs to include the xDS node id in the following cases
  - [x] Error reported to the channel (when requested resource is not found)
  - [x] Error reported from the config selector
- [x] xDS server needs to include the xDS node id when it puts the server in non-serving mode
- [x] All resource watchers' `OnResourceDoesNotExist` callback should include the xDS node id if they are throwing a log. Also, these logs should not be `Info` protected with a verbosity check, but should be `Warning` logs since resource not found is usually something that is worth sharing with the user by default. 

Solution patch:
diff --git a/xds/internal/balancer/cdsbalancer/cdsbalancer.go b/xds/internal/balancer/cdsbalancer/cdsbalancer.go
index 7aff6064..3f055a29 100644
--- a/xds/internal/balancer/cdsbalancer/cdsbalancer.go
+++ b/xds/internal/balancer/cdsbalancer/cdsbalancer.go
@@ -344,10 +344,14 @@ func (b *cdsBalancer) UpdateClientConnState(state balancer.ClientConnState) erro
 // ResolverError handles errors reported by the xdsResolver.
 func (b *cdsBalancer) ResolverError(err error) {
 	b.serializer.TrySchedule(func(context.Context) {
-		// Resource not found error is reported by the resolver when the
-		// top-level cluster resource is removed by the management server.
+		// Missing Listener or RouteConfiguration on the management server
+		// results in a 'resource not found' error from the xDS resolver. In
+		// these cases, we should stap watching all of the current clusters
+		// being watched.
 		if xdsresource.ErrType(err) == xdsresource.ErrorTypeResourceNotFound {
 			b.closeAllWatchers()
+			b.closeChildPolicyAndReportTF(err)
+			return
 		}
 		var root string
 		if b.lbCfg != nil {
@@ -372,6 +376,22 @@ func (b *cdsBalancer) closeAllWatchers() {
 	}
 }
 
+// closeChildPolicyAndReportTF closes the child policy, if it exists, and
+// updates the connectivity state of the channel to TransientFailure with an
+// error picker.
+//
+// Only executed in the context of a serializer callback.
+func (b *cdsBalancer) closeChildPolicyAndReportTF(err error) {
+	if b.childLB != nil {
+		b.childLB.Close()
+		b.childLB = nil
+	}
+	b.ccw.UpdateState(balancer.State{
+		ConnectivityState: connectivity.TransientFailure,
+		Picker:            base.NewErrPicker(err),
+	})
+}
+
 // Close cancels the CDS watch, closes the child policy and closes the
 // cdsBalancer.
 func (b *cdsBalancer) Close() {
@@ -537,16 +557,8 @@ func (b *cdsBalancer) onClusterError(name string, err error) {
 //
 // Only executed in the context of a serializer callback.
 func (b *cdsBalancer) onClusterResourceNotFound(name string) {
-	err := xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "resource name %q of type Cluster not found in received response", name)
-	if b.childLB != nil {
-		b.childLB.ResolverError(err)
-	} else {
-		// If child balancer was never created, fail the RPCs with errors.
-		b.ccw.UpdateState(balancer.State{
-			ConnectivityState: connectivity.TransientFailure,
-			Picker:            base.NewErrPicker(err),
-		})
-	}
+	err := xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "cluster %q not found", name)
+	b.closeChildPolicyAndReportTF(err)
 }
 
 // Generates discovery mechanisms for the cluster graph rooted at `name`. This
diff --git a/xds/internal/resolver/serviceconfig.go b/xds/internal/resolver/serviceconfig.go
index 7df75465..ca9f2c01 100644
--- a/xds/internal/resolver/serviceconfig.go
+++ b/xds/internal/resolver/serviceconfig.go
@@ -71,10 +71,10 @@ type xdsClusterManagerConfig struct {
 	Children map[string]xdsChildConfig `json:"children"`
 }
 
-// serviceConfigJSON produces a service config in JSON format representing all
-// the clusters referenced in activeClusters.  This includes clusters with zero
-// references, so they must be pruned first.
-func serviceConfigJSON(activeClusters map[string]*clusterInfo) ([]byte, error) {
+// serviceConfigJSON produces a service config in JSON format that contains LB
+// policy config for the "xds_cluster_manager" LB policy, with entries in the
+// children map for all active clusters.
+func serviceConfigJSON(activeClusters map[string]*clusterInfo) []byte {
 	// Generate children (all entries in activeClusters).
 	children := make(map[string]xdsChildConfig)
 	for cluster, ci := range activeClusters {
@@ -87,11 +87,13 @@ func serviceConfigJSON(activeClusters map[string]*clusterInfo) ([]byte, error) {
 		),
 	}
 
+	// This is not expected to fail as we have constructed the service config by
+	// hand right above, and therefore ok to panic.
 	bs, err := json.Marshal(sc)
 	if err != nil {
-		return nil, fmt.Errorf("failed to marshal json: %v", err)
+		panic(fmt.Sprintf("failed to marshal service config %+v: %v", sc, err))
 	}
-	return bs, nil
+	return bs
 }
 
 type virtualHost struct {
diff --git a/xds/internal/resolver/xds_resolver.go b/xds/internal/resolver/xds_resolver.go
index dfe3fcc4..c333447e 100644
--- a/xds/internal/resolver/xds_resolver.go
+++ b/xds/internal/resolver/xds_resolver.go
@@ -281,8 +281,7 @@ func (r *xdsResolver) Close() {
 // sendNewServiceConfig prunes active clusters, generates a new service config
 // based on the current set of active clusters, and sends an update to the
 // channel with that service config and the provided config selector.  Returns
-// false if an error occurs while generating the service config and the update
-// cannot be sent.
+// false if an error occurs while sending an update to the channel.
 //
 // Only executed in the context of a serializer callback.
 func (r *xdsResolver) sendNewServiceConfig(cs *configSelector) bool {
@@ -295,24 +294,28 @@ func (r *xdsResolver) sendNewServiceConfig(cs *configSelector) bool {
 		// There are no clusters and we are sending a failing configSelector.
 		// Send an empty config, which picks pick-first, with no address, and
 		// puts the ClientConn into transient failure.
-		r.cc.UpdateState(resolver.State{ServiceConfig: r.cc.ParseServiceConfig("{}")})
+		if err := r.cc.UpdateState(resolver.State{ServiceConfig: r.cc.ParseServiceConfig("{}")}); err != nil {
+			if r.logger.V(2) {
+				r.logger.Infof("Channel rejected new state (with empty service config) with error: %v", err)
+			}
+			return false
+		}
 		return true
 	}
 
-	sc, err := serviceConfigJSON(r.activeClusters)
-	if err != nil {
-		// JSON marshal error; should never happen.
-		r.logger.Errorf("For Listener resource %q and RouteConfiguration resource %q, failed to marshal newly built service config: %v", r.ldsResourceName, r.rdsResourceName, err)
-		r.cc.ReportError(err)
-		return false
-	}
+	sc := serviceConfigJSON(r.activeClusters)
 	r.logger.Infof("For Listener resource %q and RouteConfiguration resource %q, generated service config: %v", r.ldsResourceName, r.rdsResourceName, pretty.FormatJSON(sc))
 
 	// Send the update to the ClientConn.
 	state := iresolver.SetConfigSelector(resolver.State{
 		ServiceConfig: r.cc.ParseServiceConfig(string(sc)),
 	}, cs)
-	r.cc.UpdateState(xdsclient.SetClient(state, r.xdsClient))
+	if err := r.cc.UpdateState(xdsclient.SetClient(state, r.xdsClient)); err != nil {
+		if r.logger.V(2) {
+			r.logger.Infof("Channel rejected new state: %+v with error: %v", state, err)
+		}
+		return false
+	}
 	return true
 }
 
@@ -321,7 +324,7 @@ func (r *xdsResolver) sendNewServiceConfig(cs *configSelector) bool {
 // r.activeClusters for previously-unseen clusters.
 //
 // Only executed in the context of a serializer callback.
-func (r *xdsResolver) newConfigSelector() (*configSelector, error) {
+func (r *xdsResolver) newConfigSelector() *configSelector {
 	cs := &configSelector{
 		r: r,
 		virtualHost: virtualHost{
@@ -357,11 +360,7 @@ func (r *xdsResolver) newConfigSelector() (*configSelector, error) {
 		}
 		cs.routes[i].clusters = clusters
 
-		var err error
-		cs.routes[i].m, err = xdsresource.RouteToMatcher(rt)
-		if err != nil {
-			return nil, err
-		}
+		cs.routes[i].m = xdsresource.RouteToMatcher(rt)
 		cs.routes[i].actionType = rt.ActionType
 		if rt.MaxStreamDuration == nil {
 			cs.routes[i].maxStreamDuration = r.currentListener.MaxStreamDuration
@@ -381,7 +380,7 @@ func (r *xdsResolver) newConfigSelector() (*configSelector, error) {
 		atomic.AddInt32(&ci.refCount, 1)
 	}
 
-	return cs, nil
+	return cs
 }
 
 // pruneActiveClusters deletes entries in r.activeClusters with zero
@@ -437,13 +436,7 @@ func (r *xdsResolver) onResolutionComplete() {
 		return
 	}
 
-	cs, err := r.newConfigSelector()
-	if err != nil {
-		r.logger.Warningf("Failed to build a config selector for resource %q: %v", r.ldsResourceName, err)
-		r.cc.ReportError(err)
-		return
-	}
-
+	cs := r.newConfigSelector()
 	if !r.sendNewServiceConfig(cs) {
 		// JSON error creating the service config (unexpected); erase
 		// this config selector and ignore this update, continuing with
diff --git a/xds/internal/xdsclient/xdsresource/filter_chain.go b/xds/internal/xdsclient/xdsresource/filter_chain.go
index 196bb9f8..56a94dea 100644
--- a/xds/internal/xdsclient/xdsresource/filter_chain.go
+++ b/xds/internal/xdsclient/xdsresource/filter_chain.go
@@ -122,12 +122,8 @@ func (fc *FilterChain) ConstructUsableRouteConfiguration(config RouteConfigUpdat
 func (fc *FilterChain) convertVirtualHost(virtualHost *VirtualHost) (VirtualHostWithInterceptors, error) {
 	rs := make([]RouteWithInterceptors, len(virtualHost.Routes))
 	for i, r := range virtualHost.Routes {
-		var err error
 		rs[i].ActionType = r.ActionType
-		rs[i].M, err = RouteToMatcher(r)
-		if err != nil {
-			return VirtualHostWithInterceptors{}, fmt.Errorf("matcher construction: %v", err)
-		}
+		rs[i].M = RouteToMatcher(r)
 		for _, filter := range fc.HTTPFilters {
 			// Route is highest priority on server side, as there is no concept
 			// of an upstream cluster on server side.
diff --git a/xds/internal/xdsclient/xdsresource/matcher.go b/xds/internal/xdsclient/xdsresource/matcher.go
index 798f6188..cedfa586 100644
--- a/xds/internal/xdsclient/xdsresource/matcher.go
+++ b/xds/internal/xdsclient/xdsresource/matcher.go
@@ -29,7 +29,10 @@ import (
 )
 
 // RouteToMatcher converts a route to a Matcher to match incoming RPC's against.
-func RouteToMatcher(r *Route) (*CompositeMatcher, error) {
+//
+// Only expected to be called on a Route that passed validation checks by the
+// xDS client.
+func RouteToMatcher(r *Route) *CompositeMatcher {
 	var pm pathMatcher
 	switch {
 	case r.Regex != nil:
@@ -39,7 +42,7 @@ func RouteToMatcher(r *Route) (*CompositeMatcher, error) {
 	case r.Prefix != nil:
 		pm = newPathPrefixMatcher(*r.Prefix, r.CaseInsensitive)
 	default:
-		return nil, fmt.Errorf("illegal route: missing path_matcher")
+		panic("illegal route: missing path_matcher")
 	}
 
 	headerMatchers := make([]matcher.HeaderMatcher, 0, len(r.Headers))
@@ -62,7 +65,7 @@ func RouteToMatcher(r *Route) (*CompositeMatcher, error) {
 		case h.StringMatch != nil:
 			matcherT = matcher.NewHeaderStringMatcher(h.Name, *h.StringMatch, invert)
 		default:
-			return nil, fmt.Errorf("illegal route: missing header_match_specifier")
+			panic("illegal route: missing header_match_specifier")
 		}
 		headerMatchers = append(headerMatchers, matcherT)
 	}
@@ -71,7 +74,7 @@ func RouteToMatcher(r *Route) (*CompositeMatcher, error) {
 	if r.Fraction != nil {
 		fractionMatcher = newFractionMatcher(*r.Fraction)
 	}
-	return newCompositeMatcher(pm, headerMatchers, fractionMatcher), nil
+	return newCompositeMatcher(pm, headerMatchers, fractionMatcher)
 }
 
 // CompositeMatcher is a matcher that holds onto many matchers and aggregates



Test patch:
diff --git a/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go b/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
index 34f92738..f8309ab0 100644
--- a/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
+++ b/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
@@ -672,9 +672,6 @@ func (s) TestClusterUpdate_SuccessWithLRS(t *testing.T) {
 //   - when a bad cluster resource update is received after a previous good
 //     update from the management server, the cds LB policy is expected to
 //     continue using the previous good update.
-//   - when the cluster resource is removed after a previous good
-//     update from the management server, the cds LB policy is expected to put
-//     the channel in TRANSIENT_FAILURE.
 func (s) TestClusterUpdate_Failure(t *testing.T) {
 	_, resolverErrCh, _, _ := registerWrappedClusterResolverPolicy(t)
 	mgmtServer, nodeID, cc, _, _, cdsResourceRequestedCh, cdsResourceCanceledCh := setupWithManagementServer(t)
@@ -778,34 +775,6 @@ func (s) TestClusterUpdate_Failure(t *testing.T) {
 	case <-ctx.Done():
 		t.Fatal("Timeout when waiting for resolver error to be pushed to the child policy")
 	}
-
-	// Remove the cluster resource from the management server, triggering a
-	// resource-not-found error.
-	resources = e2e.UpdateOptions{
-		NodeID:         nodeID,
-		SkipValidation: true,
-	}
-	if err := mgmtServer.Update(ctx, resources); err != nil {
-		t.Fatal(err)
-	}
-
-	// Verify that the watch for the cluster resource is not cancelled.
-	sCtx, sCancel = context.WithTimeout(ctx, defaultTestShortTimeout)
-	defer sCancel()
-	select {
-	case <-sCtx.Done():
-	case <-cdsResourceCanceledCh:
-		t.Fatal("Watch for cluster resource is cancelled when not expected to")
-	}
-
-	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
-
-	// Ensure RPC fails with Unavailable. The actual error message depends on
-	// the picker returned from the priority LB policy, and therefore not
-	// checking for it here.
-	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); status.Code(err) != codes.Unavailable {
-		t.Fatalf("EmptyCall() failed with code: %v, want %v", status.Code(err), codes.Unavailable)
-	}
 }
 
 // Tests the following scenarios for resolver errors:
@@ -822,7 +791,7 @@ func (s) TestClusterUpdate_Failure(t *testing.T) {
 //     is expected to push the error down the child policy and put the channel in
 //     TRANSIENT_FAILURE. It is also expected to cancel the CDS watch.
 func (s) TestResolverError(t *testing.T) {
-	_, resolverErrCh, _, _ := registerWrappedClusterResolverPolicy(t)
+	_, resolverErrCh, _, childPolicyCloseCh := registerWrappedClusterResolverPolicy(t)
 	lis := testutils.NewListenerWrapper(t, nil)
 	mgmtServer, nodeID, cc, r, _, cdsResourceRequestedCh, cdsResourceCanceledCh := setupWithManagementServerAndListener(t, lis)
 
@@ -938,12 +907,9 @@ func (s) TestResolverError(t *testing.T) {
 
 	// Verify that the resolver error is pushed to the child policy.
 	select {
-	case err := <-resolverErrCh:
-		if err != resolverErr {
-			t.Fatalf("Error pushed to child policy is %v, want %v", err, resolverErr)
-		}
+	case <-childPolicyCloseCh:
 	case <-ctx.Done():
-		t.Fatal("Timeout when waiting for resolver error to be pushed to the child policy")
+		t.Fatal("Timeout when waiting for child policy to be closed")
 	}
 
 	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
@@ -956,6 +922,91 @@ func (s) TestResolverError(t *testing.T) {
 	}
 }
 
+// Tests scenarios involving removal of a cluster resource from the management
+// server.
+//
+//   - when the cluster resource is removed after a previous good
+//     update from the management server, the cds LB policy is expected to put
+//     the channel in TRANSIENT_FAILURE.
+//   - when the cluster resource is re-sent by the management server, RPCs
+//     should start succeeding.
+func (s) TestClusterUpdate_ResourceNotFound(t *testing.T) {
+	mgmtServer, nodeID, cc, _, _, cdsResourceRequestedCh, cdsResourceCanceledCh := setupWithManagementServer(t)
+
+	// Verify that the specified cluster resource is requested.
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	wantNames := []string{clusterName}
+	if err := waitForResourceNames(ctx, cdsResourceRequestedCh, wantNames); err != nil {
+		t.Fatal(err)
+	}
+
+	// Start a test service backend.
+	server := stubserver.StartTestService(t, nil)
+	t.Cleanup(server.Stop)
+
+	// Configure cluster and endpoints resources in the management server.
+	resources := e2e.UpdateOptions{
+		NodeID:         nodeID,
+		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)},
+		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, server.Address)})},
+		SkipValidation: true,
+	}
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	// Verify that a successful RPC can be made.
+	client := testgrpc.NewTestServiceClient(cc)
+	if _, err := client.EmptyCall(ctx, &testpb.Empty{}, grpc.WaitForReady(true)); err != nil {
+		t.Fatalf("EmptyCall() failed: %v", err)
+	}
+
+	// Remove the cluster resource from the management server, triggering a
+	// resource-not-found error.
+	resources.Clusters = nil
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	// Verify that the watch for the cluster resource is not cancelled.
+	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
+	defer sCancel()
+	select {
+	case <-sCtx.Done():
+	case <-cdsResourceCanceledCh:
+		t.Fatal("Watch for cluster resource is cancelled when not expected to")
+	}
+
+	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
+
+	// Ensure RPC fails with Unavailable.
+	wantErr := fmt.Sprintf("cluster %q not found", clusterName)
+	_, err := client.EmptyCall(ctx, &testpb.Empty{})
+	if status.Code(err) != codes.Unavailable {
+		t.Fatalf("EmptyCall() failed with code: %v, want %v", status.Code(err), codes.Unavailable)
+	}
+	if !strings.Contains(err.Error(), wantErr) {
+		t.Fatalf("EmptyCall() failed with error: %v, want %v", err, wantErr)
+	}
+
+	// Re-add the cluster resource to the management server.
+	resources = e2e.UpdateOptions{
+		NodeID:         nodeID,
+		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)},
+		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, server.Address)})},
+		SkipValidation: true,
+	}
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	// Verify that a successful RPC can be made.
+	if _, err := client.EmptyCall(ctx, &testpb.Empty{}, grpc.WaitForReady(true)); err != nil {
+		t.Fatalf("EmptyCall() failed: %v", err)
+	}
+}
+
 // Tests that closing the cds LB policy results in the the child policy being
 // closed.
 func (s) TestClose(t *testing.T) {
diff --git a/xds/internal/balancer/clusterresolver/e2e_test/balancer_test.go b/xds/internal/balancer/clusterresolver/e2e_test/balancer_test.go
index e03c497a..1689f03c 100644
--- a/xds/internal/balancer/clusterresolver/e2e_test/balancer_test.go
+++ b/xds/internal/balancer/clusterresolver/e2e_test/balancer_test.go
@@ -278,11 +278,12 @@ func (s) TestErrorFromParentLB_ResourceNotFound(t *testing.T) {
 	}
 
 	// Ensure that RPCs start to fail with expected error.
+	wantErr := fmt.Sprintf("cluster %q not found", clusterName)
 	for ; ctx.Err() == nil; <-time.After(defaultTestShortTimeout) {
 		sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
 		defer sCancel()
 		_, err := client.EmptyCall(sCtx, &testpb.Empty{})
-		if status.Code(err) == codes.Unavailable && strings.Contains(err.Error(), "all priorities are removed") {
+		if status.Code(err) == codes.Unavailable && strings.Contains(err.Error(), wantErr) {
 			break
 		}
 		if err != nil {