Repository: grpc/grpc-go
File: xds/internal/xdsclient/tests/lds_watchers_test.go (Go)

Code before:
	ldsCancel2 := xdsresource.WatchListener(client, ldsName, lw2)
	defer ldsCancel2()
	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_ResourceCaching_WithNACKError covers the case where a watch is
// registered for a resource which is already present in the cache with an old
// good update as well as latest NACK error. The test verifies that new watcher
// receives both good update and error without a new resource request being
// sent to the management server.
func (s) TestLDSWatch_ResourceCaching_NACKError(t *testing.T) {
	secondRequestReceived := grpcsync.NewEvent()

	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
		OnStreamRequest: func(id int64, req *v3discoverypb.DiscoveryRequest) error {
			// If the version is "2", it means that a second request has been
			// received (after an initial request and ack). The client should
			// not send a second request if the resource is already cached.
			if req.GetVersionInfo() == "2" {
				secondRequestReceived.Fire()
				return nil
			}
			return nil
		},
	})

	nodeID := uuid.New().String()
	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)

	// Create an xDS client with the above bootstrap contents.
	client, close, err := xdsclient.NewForTesting(xdsclient.OptionsForTesting{
		Name:     t.Name(),
		Contents: bc,
	})
	if err != nil {
		t.Fatalf("Failed to create xDS client: %v", err)
	}
	defer close()

	// Register a watch for a listener resource and have the watch
	// callback push the received update on to a channel.
	lw1 := newListenerWatcher()
	ldsCancel1 := xdsresource.WatchListener(client, ldsName, lw1)
	defer ldsCancel1()

Review comment (bug):
This is making an assumption that the management server is using monotonically increasing integers starting at `1` for version numbers. There is nothing in the xDS spec that requires management servers to that I believe. I might be wrong here.

But you could do something like what we do in `TestLDSWatch_ResourceCaching` to ensure that a second request is not sent out. There will be a second ACK here though, so you would not be able to use the exact logic from there, but you can get an idea.

Code after:
	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	if err := mgmtServer.Update(ctx, resource); err != nil {
		t.Fatalf("Failed to update management server with resource: %v, err: %v", resource, err)
	}

	// Verify the contents of the received update for existing watch.
	wantUpdate := listenerUpdateErrTuple{
		update: xdsresource.ListenerUpdate{
			RouteConfigName: rdsName,
			HTTPFilters:     []xdsresource.HTTPFilter{{Name: "router"}},
		},
	}
	if err := verifyListenerUpdate(ctx, lw1.updateCh, wantUpdate); err != nil {
		t.Fatal(err)
	}

	// Remove the listener resource on the management server.
	resource = e2e.UpdateOptions{
		NodeID:         nodeID,
		Listeners:      []*v3listenerpb.Listener{},
		SkipValidation: true,
	}
	if err := mgmtServer.Update(ctx, resource); err != nil {
		t.Fatalf("Failed to update management server with resource: %v, err: %v", resource, err)
	}

	// The existing watcher should receive a resource removed error.
	updateError := listenerUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "")}
	if err := verifyListenerUpdate(ctx, lw1.updateCh, updateError); err != nil {
		t.Fatal(err)
	}

	// New watchers attempting to register for a deleted resource should also
	// receive a "resource removed" error.
	lw2 := newListenerWatcher()
	ldsCancel2 := xdsresource.WatchListener(client, ldsName, lw2)
	defer ldsCancel2()
	if err := verifyListenerUpdate(ctx, lw2.updateCh, updateError); err != nil {
		t.Fatal(err)
	}
}

// TestLDSWatch_NACKError covers the case where an update from the management
// server is NACKed by the xdsclient. The test verifies that the error is
// propagated to the existing watcher. After NACK, if a new watcher registers
// for the resource, error is propagated to the new watcher as well.
func (s) TestLDSWatch_NACKError(t *testing.T) {
	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})

	nodeID := uuid.New().String()