xds: handle errors in xds resolver (#3651)

xds/internal/resolver/serviceconfig_test.go

@@ -48,6 +48,11 @@ const (
 		}
 	  }
 	}
+}]}`
+	testWeightedCDSNoChildJSON = `{"loadBalancingConfig":[{
+    "weighted_target_experimental": {
+	  "targets": {}
+	}
 }]}`
 )
 
@@ -62,6 +67,11 @@ func TestServiceUpdateToJSON(t *testing.T) {
 			su:       client.ServiceUpdate{WeightedCluster: map[string]uint32{testCluster1: 1}},
 			wantJSON: testClusterOnlyJSON,
 		},
+		{
+			name:     "empty weighted clusters",
+			su:       client.ServiceUpdate{WeightedCluster: nil},
+			wantJSON: testWeightedCDSNoChildJSON,
+		},
 		{
 			name: "weighted clusters",
 			su: client.ServiceUpdate{WeightedCluster: map[string]uint32{

xds/internal/resolver/xds_resolver.go

@@ -170,6 +170,18 @@ func (r *xdsResolver) run() {
 		case update := <-r.updateCh:
 			if update.err != nil {
 				r.logger.Warningf("Watch error on resource %v from xds-client %p, %v", r.target.Endpoint, r.client, update.err)
+				if xdsclient.ErrType(update.err) == xdsclient.ErrorTypeResourceNotFound {
+					// If error is resource-not-found, it means the LDS resource
+					// was removed. Send an empty service config, which picks
+					// pick-first, with no address, and puts the ClientConn into
+					// transient failure..
+					r.cc.UpdateState(resolver.State{
+						ServiceConfig: r.cc.ParseServiceConfig("{}"),
+					})
+					continue
+				}
+				// Send error to ClientConn, and balancers, if error is not
+				// resource not found.
 				r.cc.ReportError(update.err)
 				continue
 			}

xds/internal/resolver/xds_resolver_test.go

@@ -406,3 +406,47 @@ func TestXDSResolverGoodUpdateAfterError(t *testing.T) {
 		t.Fatalf("ClientConn.ReportError() received %v, want %v", gotErrVal, suErr2)
 	}
 }
+
+// TestXDSResolverResourceNotFoundError tests the cases where the resolver gets
+// a ResourceNotFoundError. It should generate a service config picking
+// weighted_target, but no child balancers.
+func TestXDSResolverResourceNotFoundError(t *testing.T) {
+	xdsC := fakeclient.NewClient()
+	xdsR, tcc, cancel := testSetup(t, setupOpts{
+		config:        &validConfig,
+		xdsClientFunc: func(_ xdsclient.Options) (xdsClientInterface, error) { return xdsC, nil },
+	})
+	defer func() {
+		cancel()
+		xdsR.Close()
+	}()
+
+	waitForWatchService(t, xdsC, targetStr)
+
+	// Invoke the watchAPI callback with a bad service update and wait for the
+	// ReportError method to be called on the ClientConn.
+	suErr := xdsclient.NewErrorf(xdsclient.ErrorTypeResourceNotFound, "resource removed error")
+	xdsC.InvokeWatchServiceCallback(xdsclient.ServiceUpdate{}, suErr)
+	if gotErrVal, gotErr := tcc.errorCh.Receive(); gotErr != testutils.ErrRecvTimeout {
+		t.Fatalf("ClientConn.ReportError() received %v, %v, want channel recv timeout", gotErrVal, gotErr)
+	}
+	gotState, err := tcc.stateCh.Receive()
+	if err != nil {
+		t.Fatalf("ClientConn.UpdateState returned error: %v", err)
+	}
+	rState := gotState.(resolver.State)
+	// This update shouldn't have xds-client in it, because it doesn't pick an
+	// xds balancer.
+	if gotClient := rState.Attributes.Value(xdsinternal.XDSClientID); gotClient != nil {
+		t.Fatalf("ClientConn.UpdateState got xdsClient: %v, want <nil>", gotClient)
+	}
+	wantParsedConfig := internal.ParseServiceConfigForTesting.(func(string) *serviceconfig.ParseResult)("{}")
+	if !internal.EqualServiceConfigForTesting(rState.ServiceConfig.Config, wantParsedConfig.Config) {
+		t.Error("ClientConn.UpdateState got wrong service config")
+		t.Errorf("gotParsed: %s", cmp.Diff(nil, rState.ServiceConfig.Config))
+		t.Errorf("wantParsed: %s", cmp.Diff(nil, wantParsedConfig.Config))
+	}
+	if err := rState.ServiceConfig.Err; err != nil {
+		t.Fatalf("ClientConn.UpdateState received error in service config: %v", rState.ServiceConfig.Err)
+	}
+}