xdsclient: include xds node ID in errors from the WatchResource API (#8093)

Author: easwars

xds/internal/xdsclient/authority.go

@@ -604,8 +604,9 @@ func (a *authority) watchResource(rType xdsresource.Type, resourceName string, w
 a.logger.Infof("New watch for type %q, resource name %q", rType.TypeName(), resourceName)
 }
 
-xdsChannel := a.xdsChannelToUse()
-if xdsChannel == nil {
+xdsChannel, err := a.xdsChannelToUse()
+if err != nil {
+a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.OnError(err, func() {}) })
 return
 }
 
@@ -739,22 +740,23 @@ func (a *authority) unwatchResource(rType xdsresource.Type, resourceName string,
 // Otherwise, it creates a new channel using the first server configuration in
 // the list of configurations, and returns that.
 //
+// A non-nil error is returned if the channel creation fails.
+//
 // Only executed in the context of a serializer callback.
-func (a *authority) xdsChannelToUse() *xdsChannelWithConfig {
+func (a *authority) xdsChannelToUse() (*xdsChannelWithConfig, error) {
 if a.activeXDSChannel != nil {
-return a.activeXDSChannel
+return a.activeXDSChannel, nil
 }
 
 sc := a.xdsChannelConfigs[0].serverConfig
 xc, cleanup, err := a.getChannelForADS(sc, a)
 if err != nil {
-a.logger.Warningf("Failed to create xDS channel: %v", err)
-return nil
+return nil, err
 }
 a.xdsChannelConfigs[0].channel = xc
 a.xdsChannelConfigs[0].cleanup = cleanup
 a.activeXDSChannel = a.xdsChannelConfigs[0]
-return a.activeXDSChannel
+return a.activeXDSChannel, nil
 }
 
 // closeXDSChannels closes all the xDS channels associated with this authority,

xds/internal/xdsclient/clientimpl_watchers.go

@@ -26,6 +26,17 @@ import (
 "google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
 )
 
+// wrappingWatcher is a wrapper around an xdsresource.ResourceWatcher that adds
+// the node ID to the error messages reported to the watcher.
+type wrappingWatcher struct {
+xdsresource.ResourceWatcher
+nodeID string
+}
+
+func (w *wrappingWatcher) OnError(err error, done xdsresource.OnDoneFunc) {
+w.ResourceWatcher.OnError(fmt.Errorf("[xDS node id: %v]: %v", w.nodeID, err), done)
+}
+
 // WatchResource uses xDS to discover the resource associated with the provided
 // resource name. The resource type implementation determines how xDS responses
 // are are deserialized and validated, as received from the xDS management
@@ -43,6 +54,11 @@ func (c *clientImpl) WatchResource(rType xdsresource.Type, resourceName string,
 return func() {}
 }
 
+watcher = &wrappingWatcher{
+ResourceWatcher: watcher,
+nodeID: c.config.Node().GetId(),
+}
+
 if err := c.resourceTypes.maybeRegister(rType); err != nil {
 logger.Warningf("Watch registered for name %q of type %q which is already registered", rType.TypeName(), resourceName)
 c.serializer.TrySchedule(func(context.Context) { watcher.OnError(err, func() {}) })
@@ -53,9 +69,7 @@ func (c *clientImpl) WatchResource(rType xdsresource.Type, resourceName string,
 a := c.getAuthorityForResource(n)
 if a == nil {
 logger.Warningf("Watch registered for name %q of type %q, authority %q is not found", rType.TypeName(), resourceName, n.Authority)
-c.serializer.TrySchedule(func(context.Context) {
-watcher.OnError(fmt.Errorf("authority %q not found in bootstrap config for resource %q", n.Authority, resourceName), func() {})
-})
+watcher.OnError(fmt.Errorf("authority %q not found in bootstrap config for resource %q", n.Authority, resourceName), func() {})
 return func() {}
 }
 // The watchResource method on the authority is invoked with n.String()

xds/internal/xdsclient/tests/misc_watchers_test.go

@@ -22,16 +22,19 @@ import (
 "context"
 "encoding/json"
 "fmt"
+"strings"
 "testing"
 
 "github.com/google/uuid"
+"google.golang.org/grpc"
 "google.golang.org/grpc/internal/testutils"
 "google.golang.org/grpc/internal/testutils/xds/e2e"
 "google.golang.org/grpc/internal/testutils/xds/fakeserver"
 "google.golang.org/grpc/internal/xds/bootstrap"
 "google.golang.org/grpc/xds/internal"
 xdstestutils "google.golang.org/grpc/xds/internal/testutils"
 "google.golang.org/grpc/xds/internal/xdsclient"
+xdsclientinternal "google.golang.org/grpc/xds/internal/xdsclient/internal"
 "google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
 "google.golang.org/grpc/xds/internal/xdsclient/xdsresource/version"
 "google.golang.org/protobuf/types/known/anypb"
@@ -371,3 +374,135 @@ func readDiscoveryResponseAndCheckForNonEmptyNodeProto(ctx context.Context, reqC
 }
 return nil
 }
+
+type testRouteConfigResourceType struct{}
+
+func (testRouteConfigResourceType) TypeURL() string { return version.V3RouteConfigURL }
+func (testRouteConfigResourceType) TypeName() string { return "RouteConfigResource" }
+func (testRouteConfigResourceType) AllResourcesRequiredInSotW() bool { return false }
+func (testRouteConfigResourceType) Decode(*xdsresource.DecodeOptions, *anypb.Any) (*xdsresource.DecodeResult, error) {
+return nil, nil
+}
+
+// Tests that the errors returned by the xDS client when watching a resource
+// contain the node ID that was used to create the client. This test covers two
+// scenarios:
+//
+// 1. When a watch is registered for an already registered resource type, but
+// this time with a different implementation,
+// 2. When a watch is registered for a resource name whose authority is not
+// found in the bootstrap configuration.
+func (s) TestWatchErrorsContainNodeID(t *testing.T) {
+mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
+
+// Create bootstrap configuration pointing to the above management server.
+nodeID := uuid.New().String()
+bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
+
+// Create an xDS client with the above bootstrap contents.
+config, err := bootstrap.NewConfigFromContents(bc)
+if err != nil {
+t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bc), err)
+}
+pool := xdsclient.NewPool(config)
+client, close, err := pool.NewClientForTesting(xdsclient.OptionsForTesting{
+Name: t.Name(),
+})
+if err != nil {
+t.Fatalf("Failed to create xDS client: %v", err)
+}
+defer close()
+
+ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+defer cancel()
+
+t.Run("Multiple_ResourceType_Implementations", func(t *testing.T) {
+const routeConfigName = "route-config-name"
+watcher := xdstestutils.NewTestResourceWatcher()
+client.WatchResource(routeConfigResourceType, routeConfigName, watcher)
+
+sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
+defer sCancel()
+select {
+case <-sCtx.Done():
+case <-watcher.UpdateCh:
+t.Fatal("Unexpected resource update")
+case <-watcher.ErrorCh:
+t.Fatal("Unexpected resource error")
+case <-watcher.ResourceDoesNotExistCh:
+t.Fatal("Unexpected resource does not exist")
+}
+
+client.WatchResource(testRouteConfigResourceType{}, routeConfigName, watcher)
+select {
+case <-ctx.Done():
+t.Fatal("Timeout when waiting for error callback to be invoked")
+case err := <-watcher.ErrorCh:
+if err == nil || !strings.Contains(err.Error(), nodeID) {
+t.Fatalf("Unexpected error: %v, want error with node ID: %q", err, nodeID)
+}
+}
+})
+
+t.Run("Missing_Authority", func(t *testing.T) {
+const routeConfigName = "xdstp://nonexistant-authority/envoy.config.route.v3.RouteConfiguration/route-config-name"
+watcher := xdstestutils.NewTestResourceWatcher()
+client.WatchResource(routeConfigResourceType, routeConfigName, watcher)
+
+select {
+case <-ctx.Done():
+t.Fatal("Timeout when waiting for error callback to be invoked")
+case err := <-watcher.ErrorCh:
+if err == nil || !strings.Contains(err.Error(), nodeID) {
+t.Fatalf("Unexpected error: %v, want error with node ID: %q", err, nodeID)
+}
+}
+})
+}
+
+// Tests that the errors returned by the xDS client when watching a resource
+// contain the node ID when channel creation to the management server fails.
+func (s) TestWatchErrorsContainNodeID_ChannelCreationFailure(t *testing.T) {
+mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
+
+// Create bootstrap configuration pointing to the above management server.
+nodeID := uuid.New().String()
+bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
+
+// Create an xDS client with the above bootstrap contents.
+config, err := bootstrap.NewConfigFromContents(bc)
+if err != nil {
+t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bc), err)
+}
+pool := xdsclient.NewPool(config)
+client, close, err := pool.NewClientForTesting(xdsclient.OptionsForTesting{
+Name: t.Name(),
+})
+if err != nil {
+t.Fatalf("Failed to create xDS client: %v", err)
+}
+defer close()
+
+ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+defer cancel()
+
+// Override the xDS channel dialer with one that always fails.
+origDialer := xdsclientinternal.GRPCNewClient
+xdsclientinternal.GRPCNewClient = func(target string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
+return nil, fmt.Errorf("failed to create channel")
+}
+defer func() { xdsclientinternal.GRPCNewClient = origDialer }()
+
+const routeConfigName = "route-config-name"
+watcher := xdstestutils.NewTestResourceWatcher()
+client.WatchResource(routeConfigResourceType, routeConfigName, watcher)
+
+select {
+case <-ctx.Done():
+t.Fatal("Timeout when waiting for error callback to be invoked")
+case err := <-watcher.ErrorCh:
+if err == nil || !strings.Contains(err.Error(), nodeID) {
+t.Fatalf("Unexpected error: %v, want error with node ID: %q", err, nodeID)
+}
+}
+}