test to verify both update and error are sent

Author: purnesh42H

xds/internal/xdsclient/authority.go

@@ -633,13 +633,19 @@ func (a *authority) watchResource(rType xdsresource.Type, resourceName string, w
 		// Always add the new watcher to the set of watchers.
 		state.watchers[watcher] = true
 
-		// If we have a cached copy of the resource, notify the new watcher.
+		// If we have a cached copy of the resource, notify the new watcher
+		// immediately.
 		if state.cache != nil {
 			if a.logger.V(2) {
 				a.logger.Infof("Resource type %q with resource name %q found in cache: %s", rType.TypeName(), resourceName, state.cache.ToJSON())
 			}
 			resource := state.cache
 			a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.OnUpdate(resource, func() {}) })
+			// If last update was NACK'd, notify the new watcher of error
+			// immediately as well.
+			if state.md.Status == xdsresource.ServiceStatusNACKed && state.md.ErrState != nil {
+				a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.OnError(state.md.ErrState.Err, func() {}) })
+			}
 		}
 		cleanup = a.unwatchResource(rType, resourceName, watcher)
 	}, func() {

xds/internal/xdsclient/tests/lds_watchers_test.go

@@ -90,6 +90,33 @@ func (cw *listenerWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc)
 	onDone()
 }
 
+type listenerWatcherMultiple struct {
+	updateCh *testutils.Channel
+}
+
+func newListenerWatcherMultiple() *listenerWatcherMultiple {
+	return &listenerWatcherMultiple{updateCh: testutils.NewChannelWithSize(2)}
+}
+
+func (cw *listenerWatcherMultiple) OnUpdate(update *xdsresource.ListenerResourceData, onDone xdsresource.OnDoneFunc) {
+	cw.updateCh.Send(listenerUpdateErrTuple{update: update.Resource})
+	onDone()
+}
+
+func (cw *listenerWatcherMultiple) OnError(err error, onDone xdsresource.OnDoneFunc) {
+	// When used with a go-control-plane management server that continuously
+	// resends resources which are NACKed by the xDS client, using a `Replace()`
+	// here and in OnResourceDoesNotExist() simplifies tests which will have
+	// access to the most recently received error.
+	cw.updateCh.Send(listenerUpdateErrTuple{err: err})
+	onDone()
+}
+
+func (cw *listenerWatcherMultiple) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
+	cw.updateCh.Send(listenerUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "Listener not found in received response")})
+	onDone()
+}
+
 // badListenerResource returns a listener resource for the given name which does
 // not contain the `RouteSpecifier` field in the HTTPConnectionManager, and
 // hence is expected to be NACKed by the client.
@@ -547,7 +574,7 @@ func (s) TestLDSWatch_ThreeWatchesForDifferentResourceNames(t *testing.T) {
 // a resource which is already present in the cache.  The test verifies that the
 // watch callback is invoked with the contents from the cache, instead of a
 // request being sent to the management server.
-func (s) TestLDSWatch_ResourceCaching(t *testing.T) {
+func TestLDSWatch_ResourceCaching(t *testing.T) {
 	firstRequestReceived := false
 	firstAckReceived := grpcsync.NewEvent()
 	secondRequestReceived := grpcsync.NewEvent()
@@ -926,7 +953,7 @@ func (s) TestLDSWatch_NACKError(t *testing.T) {
 // good update and latest NACK error. The test verifies that new watcher
 // receives both good update and error without request being sent to the
 // management server.
-func (s) TestLDSWatch_ResourceCaching_NACKError(t *testing.T) {
+func TestLDSWatch_ResourceCaching_NACKError(t *testing.T) {
 	firstRequestReceived := false
 	firstAckReceived := grpcsync.NewEvent()
 	secondRequestReceived := grpcsync.NewEvent()
@@ -943,6 +970,13 @@ func (s) TestLDSWatch_ResourceCaching_NACKError(t *testing.T) {
 				firstAckReceived.Fire()
 				return nil
 			}
+			// If the request version remains "1" while the nonce keeps
+			// increasing, it indicates the client is repeatedly NACKing
+			// updates from the server but not sending any new resource
+			// request.
+			if req.GetVersionInfo() == "1" {
+				return nil
+			}
 			// Any requests after the first request and ack, are not expected.
 			secondRequestReceived.Fire()
 			return nil
@@ -1018,21 +1052,12 @@ func (s) TestLDSWatch_ResourceCaching_NACKError(t *testing.T) {
 
 	// Register another watch for the same resource. This should get the update
 	// and error from the cache.
-	lw2 := newListenerWatcher()
+	lw2 := newListenerWatcherMultiple()
 	ldsCancel2 := xdsresource.WatchListener(client, ldsName, lw2)
 	defer ldsCancel2()
 	if err := verifyListenerUpdate(ctx, lw2.updateCh, wantUpdate); err != nil {
 		t.Fatal(err)
 	}
-	u, err = lw2.updateCh.Receive(ctx)
-	if err != nil {
-		t.Fatalf("timeout when waiting for a listener resource from the management server: %v", err)
-	}
-	gotErr = u.(listenerUpdateErrTuple).err
-	if gotErr == nil || !strings.Contains(gotErr.Error(), wantListenerNACKErr) {
-		t.Fatalf("update received with error: %v, want %q", gotErr, wantListenerNACKErr)
-	}
-
 	// No request should get sent out as part of this watch.
 	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
 	defer sCancel()
@@ -1041,6 +1066,14 @@ func (s) TestLDSWatch_ResourceCaching_NACKError(t *testing.T) {
 	case <-secondRequestReceived.Done():
 		t.Fatal("xdsClient sent out request instead of using update from cache")
 	}
+	u, err = lw2.updateCh.Receive(ctx)
+	if err != nil {
+		t.Fatalf("timeout when waiting for a listener resource from the management server: %v", err)
+	}
+	gotErr = u.(listenerUpdateErrTuple).err
+	if gotErr == nil || !strings.Contains(gotErr.Error(), wantListenerNACKErr) {
+		t.Fatalf("update received with error: %v, want %q", gotErr, wantListenerNACKErr)
+	}
 }
 
 // TestLDSWatch_PartialValid covers the case where a response from the