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

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

	// Verify that the expected error is propagated to the watcher.
	u, err := lw.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)
	}

	// Verify that the expected error is propagated to the new watcher as well.
	lw2 := newListenerWatcher()
	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
		},
	})


Review comment (suggestion):
The error type in the `xdsresource` package provides a way to get to the underlying error type. See: https://github.com/grpc/grpc-go/blob/master/xds/internal/xdsclient/xdsresource/errors.go#L61

I think we should use that instead of string comparison.

I agree existing tests are not doing that. But that is not a good enough reason to make new code not do that. In fact, it would be wonderful in existing code could be cleaned up as well.

Code after:
	if err != nil {
		t.Fatalf("Failed to create xDS client: %v", err)
	}
	defer close()

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

	// Configure the management server to return listener resource,
	// corresponding to the registered watch.
	resource := e2e.UpdateOptions{
		NodeID:         nodeID,
		Listeners:      []*v3listenerpb.Listener{e2e.DefaultClientListener(ldsName, rdsName)},
		SkipValidation: true,
	}
	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)
	}